fix(migration): complete-invoice-lines cron visits registers smallest first (#2341)

* fix(migration): complete-invoice-lines cron visits registers smallest first The hourly pass ordered its work by consent recency, which says nothing about work size: a 1 125-invoice register on the newest consent used two runs in a row while a 384-invoice register three consents older was skipped for budget both times. Each run now sizes every usable consent's register on our side first (one indexed HEAD count of the non-draft invoices without rows, no provider call) and then hands the registers with anything left to the pass smallest first, each within its share of the run. Shortest job first: a register that fits its share is done this run whatever was accepted after it; the one that needs several runs takes what is left of each. Nothing is stored between runs, and the budget constants are unchanged. What a run does not reach is by construction its largest registers; they are logged and returned as `deferred` with their counts so a register that is deferred hour after hour is visible. The count is proven against a real PostgREST (tool-pg) because `invoice_items=is.null` on a to-many embed is resolved there, not in Postgres or the type system. Closes #2309 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SaJfqNi4VmsG8FMKq99G6 * test(schema): teach the phantom-column guard PostgREST's embed-null filter The static guard read `.is('invoice_items', null)` as a column of `invoices` and failed CI on #2341. PostgREST's null filter on an embedded resource (`?invoice_items=is.null`, the anti-join on a to-many embed: the parents whose embed is empty) names the embed declared in the same chain's select, not a column. The scanner already registers every embed alias per chain for dotted filters; a bare name that is a registered embed, used with `is` (or `not` / `filter` with the `is` operator, the only operators that reach an embed), is now recognised and checked no further. Any other operator on a bare embed name, and `is` on a name the select never embedded, are still accused, with cases for both. The grammar itself is proven on a real PostgREST by complete-invoice-lines-count.tool.test.ts. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SaJfqNi4VmsG8FMKq99G6 --------- Co-authored-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-06 18:38:31 +02:00
parent b71f2bf425
commit 8313f527c9
7 changed files with 525 additions and 62 deletions
@@ -1,11 +1,12 @@
-import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
 import { NextResponse } from 'next/server'

 /**
 * The hourly pass that fills in the rows the migration's bounded hydration
 * did not reach. The route is thin: refuse without the cron secret, refuse
- * when the extension is off, then hand each recent accepted consent to the
- * pass with its share of the run and add the counts up.
+ * when the extension is off, size every usable consent's register on our
+ * side, then hand the registers with anything left to the pass smallest
+ * first, each with its share of the run, and add the counts up.
 */

 vi.mock('@/lib/extensions/loader', () => ({ loadExtensions: vi.fn() }))
@@ -31,16 +32,21 @@ vi.mock('@/lib/auth/api-keys', () => ({

 vi.mock('@/extensions/general/arcim-migration/lib/complete-invoice-lines', () => ({
  completeMigratedInvoiceLines: vi.fn(),
+  countRowlessInvoices: vi.fn(),
 }))

 import { GET, maxDuration, consentIsUsable } from '../route'
 import { extensionRegistry } from '@/lib/extensions/registry'
 import { verifyCronSecret } from '@/lib/auth/cron'
-import { completeMigratedInvoiceLines } from '@/extensions/general/arcim-migration/lib/complete-invoice-lines'
+import {
+  completeMigratedInvoiceLines,
+  countRowlessInvoices,
+} from '@/extensions/general/arcim-migration/lib/complete-invoice-lines'

 const mockRegistryGet = vi.mocked(extensionRegistry.get)
 const mockVerifyCronSecret = vi.mocked(verifyCronSecret)
 const mockComplete = vi.mocked(completeMigratedInvoiceLines)
+const mockCount = vi.mocked(countRowlessInvoices)

 const EMPTY = {
  candidates: 0, providerInvoices: 0, matched: 0, unmatched: 0, completed: 0, headersUpdated: 0,
@@ -49,6 +55,8 @@ const EMPTY = {
 }

 const DAY_MS = 24 * 60 * 60 * 1000
+const RUN_BUDGET_MS = 240_000
+const PER_COMPANY_BUDGET_MS = 120_000

 /** A consent whose access token expired `daysAgo` days ago (null: never expires). */
 function consent(id: string, companyId: string, daysAgo: number | null, provider = 'fortnox') {
@@ -63,19 +71,45 @@ function consent(id: string, companyId: string, daysAgo: number | null, provider
  }
 }

+/** Row-less invoices per company, as the sizing phase finds them. */
+function sizes(byCompany: Record<string, number>) {
+  mockCount.mockImplementation(async (_supabase, companyId) => byCompany[companyId] ?? 0)
+}
+
+/** A pass that completes everything it is given and reports the size it found. */
+function completesAll(byCompany: Record<string, number>) {
+  mockComplete.mockImplementation(async ({ companyId }) => {
+    const n = byCompany[companyId] ?? 0
+    return { ...EMPTY, candidates: n, matched: n, completed: n, headersUpdated: n }
+  })
+}
+
 function makeRequest() {
  return new Request('http://localhost/api/extensions/arcim-migration/complete-invoice-lines/cron', {
    headers: { authorization: 'Bearer synthetic-cron-secret' },
  })
 }

+/** Freeze the clock so a share is exactly its constant and the run's spend is what the pass says it spent. */
+function freezeClock() {
+  vi.useFakeTimers({ toFake: ['Date'] })
+  vi.setSystemTime(new Date('2026-09-06T10:20:00Z'))
+}
+
 beforeEach(() => {
  vi.clearAllMocks()
+  mockCount.mockReset()
+  mockComplete.mockReset()
  mockVerifyCronSecret.mockReturnValue(null)
  mockRegistryGet.mockReturnValue({ id: 'arcim-migration' } as never)
+  mockCount.mockResolvedValue(0)
  h.consents = { data: [], error: null }
 })

+afterEach(() => {
+  vi.useRealTimers()
+})
+
 describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () => {
  it('reserves the full function window: hydration is rate-limited at the provider', () => {
    expect(maxDuration).toBe(300)
@@ -87,6 +121,7 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =
    const response = await GET(makeRequest())

    expect(response.status).toBe(401)
+    expect(mockCount).not.toHaveBeenCalled()
    expect(mockComplete).not.toHaveBeenCalled()
  })

@@ -98,10 +133,48 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =

    expect(response.status).toBe(503)
    expect(body.code).toBe('EXTENSION_DISABLED')
+    expect(mockCount).not.toHaveBeenCalled()
    expect(mockComplete).not.toHaveBeenCalled()
  })

-  it('runs the pass once per usable consent and adds the counts up', async () => {
+  it('sizes every usable register before the first provider call, then completes them smallest first', async () => {
+    // 2026-09-05 on prod: the 1 125-invoice register sat on the newest
+    // consent and used two runs in a row; the 384-invoice one three consents
+    // older waited both times. Consent age says nothing about work size.
+    h.consents = {
+      data: [
+        consent('c-newest', 'co-big', 0),
+        consent('c-mid', 'co-small', 1),
+        consent('c-oldest', 'co-medium', 3),
+      ],
+      error: null,
+    }
+    const work = { 'co-big': 1125, 'co-small': 155, 'co-medium': 384 }
+    sizes(work)
+    completesAll(work)
+
+    const response = await GET(makeRequest())
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(mockCount.mock.calls.map((c) => c[1])).toEqual(['co-big', 'co-small', 'co-medium'])
+    expect(mockComplete.mock.calls.map((c) => c[0].companyId)).toEqual(['co-small', 'co-medium', 'co-big'])
+    expect(mockComplete.mock.calls.map((c) => c[0].consentId)).toEqual(['c-mid', 'c-oldest', 'c-newest'])
+    // Sizing is a separate phase: the last count lands before the first pass.
+    expect(Math.max(...mockCount.mock.invocationCallOrder)).toBeLessThan(Math.min(...mockComplete.mock.invocationCallOrder))
+    expect(body.data).toMatchObject({
+      consents: 3,
+      consentsStale: 0,
+      consentsFailed: 0,
+      companies: 3,
+      candidates: 1664,
+      completed: 1664,
+      skippedForBudget: 0,
+      deferred: [],
+    })
+  })
+
+  it('adds the counts up across the registers it completed', async () => {
    h.consents = {
      data: [
        consent('c-new', 'co-1', 0),
@@ -110,20 +183,20 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =
      ],
      error: null,
    }
+    sizes({ 'co-1': 311, 'co-2': 384, 'co-3': 0 })
    mockComplete
      .mockResolvedValueOnce({ ...EMPTY, candidates: 311, matched: 311, completed: 300, headersUpdated: 300, notHydrated: 11, remaining: 11 })
      .mockResolvedValueOnce({ ...EMPTY, candidates: 384, matched: 384, completed: 384, headersUpdated: 358 })
-      .mockResolvedValueOnce({ ...EMPTY })

    const response = await GET(makeRequest())
    const body = await response.json()

    expect(response.status).toBe(200)
-    expect(mockComplete).toHaveBeenCalledTimes(3)
+    expect(mockComplete).toHaveBeenCalledTimes(2)
    expect(mockComplete.mock.calls[0][0]).toMatchObject({ companyId: 'co-1', consentId: 'c-new' })
    expect(mockComplete.mock.calls[0][0].budgetMs).toBeGreaterThan(0)
-    expect(mockComplete.mock.calls[0][0].budgetMs).toBeLessThanOrEqual(120_000)
-    // The company with nothing to complete is not counted as worked on.
+    expect(mockComplete.mock.calls[0][0].budgetMs).toBeLessThanOrEqual(PER_COMPANY_BUDGET_MS)
+    // The company with nothing to complete is sized, never passed, not counted.
    expect(body.data).toMatchObject({
      consents: 3,
      consentsStale: 0,
@@ -138,11 +211,115 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =
    })
  })

-  it('isolates a failing consent: the others still run and the failure is counted', async () => {
+  it('skips a register with nothing left: no pass, no provider contact, no budget spent', async () => {
+    freezeClock()
+    h.consents = {
+      data: [consent('c-done', 'co-done', 0), consent('c-todo', 'co-todo', 1)],
+      error: null,
+    }
+    sizes({ 'co-done': 0, 'co-todo': 12 })
+    completesAll({ 'co-todo': 12 })
+
+    const response = await GET(makeRequest())
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(mockCount).toHaveBeenCalledTimes(2)
+    expect(mockComplete).toHaveBeenCalledTimes(1)
+    // The register behind it still gets a whole share: sizing cost it nothing.
+    expect(mockComplete.mock.calls[0][0]).toMatchObject({
+      companyId: 'co-todo', consentId: 'c-todo', budgetMs: PER_COMPANY_BUDGET_MS,
+    })
+    expect(body.data).toMatchObject({ consents: 2, companies: 1, candidates: 12, completed: 12, skippedForBudget: 0 })
+  })
+
+  it('stops at the run deadline and defers what it did not reach, which is the largest registers', async () => {
+    freezeClock()
+    h.consents = {
+      data: [
+        consent('c-1', 'co-big', 0),
+        consent('c-2', 'co-small', 1),
+        consent('c-3', 'co-medium', 2),
+      ],
+      error: null,
+    }
+    const work = { 'co-big': 1125, 'co-small': 155, 'co-medium': 384 }
+    sizes(work)
+    // Each register uses the whole share it is given: two shares fill the run.
+    mockComplete.mockImplementation(async ({ companyId, budgetMs }) => {
+      vi.setSystemTime(Date.now() + (budgetMs ?? 0))
+      const n = work[companyId as keyof typeof work]
+      return { ...EMPTY, candidates: n, matched: n, completed: n, headersUpdated: n }
+    })
+
+    const response = await GET(makeRequest())
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(mockComplete.mock.calls.map((c) => c[0].companyId)).toEqual(['co-small', 'co-medium'])
+    expect(mockComplete.mock.calls.map((c) => c[0].budgetMs)).toEqual([PER_COMPANY_BUDGET_MS, RUN_BUDGET_MS - PER_COMPANY_BUDGET_MS])
+    expect(body.data).toMatchObject({
+      consents: 3,
+      companies: 2,
+      candidates: 539,
+      skippedForBudget: 1,
+      deferred: [{ companyId: 'co-big', candidates: 1125 }],
+    })
+  })
+
+  it('gives a register reached late the remainder of the run, not a whole share', async () => {
+    freezeClock()
+    h.consents = {
+      data: [consent('c-1', 'co-a', 0), consent('c-2', 'co-b', 1), consent('c-3', 'co-c', 2)],
+      error: null,
+    }
+    const work = { 'co-a': 10, 'co-b': 20, 'co-c': 30 }
+    sizes(work)
+    // 120 s + 100 s spent: 20 s remain, exactly the minimum worth starting on.
+    const spend: Record<string, number> = { 'co-a': 120_000, 'co-b': 100_000, 'co-c': 5_000 }
+    mockComplete.mockImplementation(async ({ companyId }) => {
+      vi.setSystemTime(Date.now() + spend[companyId])
+      const n = work[companyId as keyof typeof work]
+      return { ...EMPTY, candidates: n, matched: n, completed: n }
+    })
+
+    const response = await GET(makeRequest())
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(mockComplete.mock.calls.map((c) => [c[0].companyId, c[0].budgetMs])).toEqual([
+      ['co-a', PER_COMPANY_BUDGET_MS],
+      ['co-b', PER_COMPANY_BUDGET_MS],
+      ['co-c', 20_000],
+    ])
+    expect(body.data).toMatchObject({ companies: 3, skippedForBudget: 0 })
+  })
+
+  it('isolates a failing count: that consent is reported failed and the others still run', async () => {
+    h.consents = {
+      data: [consent('c-broken', 'co-broken', 0), consent('c-live', 'co-live', 1)],
+      error: null,
+    }
+    mockCount.mockImplementation(async (_supabase, companyId) => {
+      if (companyId === 'co-broken') throw new Error('canceling statement due to statement timeout')
+      return 7
+    })
+    completesAll({ 'co-live': 7 })
+
+    const response = await GET(makeRequest())
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(mockComplete.mock.calls.map((c) => c[0].companyId)).toEqual(['co-live'])
+    expect(body.data).toMatchObject({ consents: 2, consentsFailed: 1, companies: 1, completed: 7 })
+  })
+
+  it('isolates a failing pass: the others still run and the failure is counted', async () => {
    h.consents = {
      data: [consent('c-revoked', 'co-1', 1), consent('c-live', 'co-2', 2)],
      error: null,
    }
+    sizes({ 'co-1': 5, 'co-2': 5 })
    mockComplete
      .mockRejectedValueOnce(new Error('Token refresh failed for fortnox; the connection must be re-authorized'))
      .mockResolvedValueOnce({ ...EMPTY, candidates: 5, matched: 5, completed: 5, headersUpdated: 5 })
@@ -152,6 +329,8 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =

    expect(response.status).toBe(200)
    expect(mockComplete).toHaveBeenCalledTimes(2)
+    // Equal sizes keep the scan order: the newer consent first.
+    expect(mockComplete.mock.calls.map((c) => c[0].consentId)).toEqual(['c-revoked', 'c-live'])
    expect(body.data).toMatchObject({ consents: 2, consentsFailed: 1, companies: 1, completed: 5 })
  })

@@ -169,17 +348,20 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =
      ],
      error: null,
    }
-    mockComplete.mockResolvedValue({ ...EMPTY })
+    sizes({ 'co-1': 1, 'co-2': 1, 'co-3': 1, 'co-4': 1 })
+    completesAll({ 'co-2': 1, 'co-3': 1 })

    const response = await GET(makeRequest())
    const body = await response.json()

    expect(response.status).toBe(200)
+    // A dead consent is not even sized: nothing about it can be acted on.
+    expect(mockCount.mock.calls.map((c) => c[1])).toEqual(['co-2', 'co-3'])
    expect(mockComplete.mock.calls.map((c) => c[0].consentId)).toEqual(['c-live', 'c-bokio'])
    expect(body.data).toMatchObject({ consents: 2, consentsStale: 2 })
  })

-  it('does not page: every usable consent is visited, not only the newest few', async () => {
+  it('does not page: every usable consent is sized, not only the newest few', async () => {
    // 57 consents were accepted in the last 60 days on prod (2026-09-05); a
    // fixed page of the newest ones would leave older companies with row-less
    // invoices waiting forever behind companies that are already done.
@@ -187,14 +369,16 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =
      data: Array.from({ length: 80 }, (_, i) => consent(`c-${i}`, `co-${i}`, 1)),
      error: null,
    }
-    mockComplete.mockResolvedValue({ ...EMPTY })
+    mockCount.mockResolvedValue(1)
+    mockComplete.mockResolvedValue({ ...EMPTY, candidates: 1, matched: 1, completed: 1 })

    const response = await GET(makeRequest())
    const body = await response.json()

    expect(response.status).toBe(200)
+    expect(mockCount).toHaveBeenCalledTimes(80)
    expect(mockComplete).toHaveBeenCalledTimes(80)
-    expect(body.data).toMatchObject({ consents: 80, skippedForBudget: 0 })
+    expect(body.data).toMatchObject({ consents: 80, companies: 80, skippedForBudget: 0 })
  })

  it('consentIsUsable reads the token row, tolerating either embed cardinality', () => {
@@ -214,6 +398,7 @@ describe('GET /api/extensions/arcim-migration/complete-invoice-lines/cron', () =
    const response = await GET(makeRequest())

    expect(response.status).toBeGreaterThanOrEqual(500)
+    expect(mockCount).not.toHaveBeenCalled()
    expect(mockComplete).not.toHaveBeenCalled()
  })
 })
@@ -5,6 +5,7 @@ import { withCronContext } from '@/lib/api/with-cron-context'
 import { createServiceClientNoCookies } from '@/lib/auth/api-keys'
 import {
  completeMigratedInvoiceLines,
+  countRowlessInvoices,
  type CompleteInvoiceLinesResult,
 } from '@/extensions/general/arcim-migration/lib/complete-invoice-lines'

@@ -15,14 +16,27 @@ import {
 *
 * The migration hydrates the provider's detail form inside a fixed budget
 * and reports the shortfall; this is what picks the shortfall up. Every run
- * walks the accepted consents whose credentials can still be used, newest
- * first, and for each company completes as many of its row-less invoices as
- * its share of the run allows. A company with nothing left costs one query
- * and no provider call (the pass checks our side before it touches the
- * consent), so walking every live consent is cheap and no company waits
- * behind a fixed page of newer ones. Scheduled hourly in vercel.json (and
- * the Docker crontabs); a company the size of Clearstoq (1 125 invoices) is
- * done after two or three runs.
+ * walks the accepted consents whose credentials can still be used, in two
+ * phases. First it sizes each register on our side: one indexed count of the
+ * invoices still without rows per consent, no provider call. Then it hands
+ * the registers with anything left to the pass smallest first, each within
+ * its share of the run. Shortest job first: a register that fits its share
+ * is finished this run whatever was accepted after it, and a register that
+ * needs several runs takes what is left of each instead of pushing every
+ * older company back by an hour per run (on 2026-09-05 a 1 125-invoice
+ * register on the newest consent used two runs in a row while a 384-invoice
+ * register three consents older, about 100 s of work, was skipped for budget
+ * both times). Nothing is stored between runs: the counts are taken fresh
+ * every hour, so a register the wizard or an earlier run finished simply
+ * stops appearing.
+ *
+ * What a run does not reach is, by construction, its largest registers. They
+ * are reported as deferred with their counts in the run summary and in the
+ * response, so a register that is deferred hour after hour (possible only
+ * while smaller registers keep arriving faster than the run clears them) is
+ * visible rather than silent. Scheduled hourly in vercel.json (and the
+ * Docker crontabs); a company the size of Clearstoq (1 125 invoices) is done
+ * after two or three runs.
 *
 * "Can still be used" is read off the token row, not the consent's age:
 * Fortnox issues a new refresh token on every refresh and each one lives 45
@@ -40,7 +54,7 @@ export const maxDuration = 300
 const RUN_BUDGET_MS = 240_000
 /** One company's share of provider detail fetches per run. */
 const PER_COMPANY_BUDGET_MS = 120_000
-/** Below this the remaining companies wait for the next run. */
+/** Below this the registers still in line (the largest ones) wait for the next run. */
 const MIN_COMPANY_BUDGET_MS = 20_000
 /**
 * A token pair not refreshed for this long cannot be refreshed any more
@@ -58,6 +72,12 @@ interface ConsentRow {
  provider_consent_tokens: { token_expires_at: string | null } | { token_expires_at: string | null }[] | null
 }

+/** A usable consent whose company still has invoices without rows, sized by that count. */
+interface Register {
+  consent: ConsentRow
+  candidates: number
+}
+
 /** The consent's token row, whichever cardinality PostgREST rendered it with. */
 function tokenOf(consent: ConsentRow): { token_expires_at: string | null } | null {
  const tokens = consent.provider_consent_tokens
@@ -96,6 +116,8 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines'
    .select('id, company_id, provider, created_at, provider_consent_tokens(token_expires_at)')
    .eq('status', 1)
    .not('provider', 'is', null)
+    // A stable scan order under the cap, and the tiebreak between registers
+    // of equal size (the sort below is stable): the newer consent goes first.
    .order('created_at', { ascending: false })
    .limit(MAX_CONSENTS_SCANNED)

@@ -105,9 +127,28 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines'

  const now = Date.now()
  const scanned = (data ?? []) as ConsentRow[]
-  const consents = scanned.filter((consent) => consentIsUsable(consent, now))
+  const usable = scanned.filter((consent) => consentIsUsable(consent, now))
  const deadline = now + RUN_BUDGET_MS
-  let skippedForBudget = 0
+
+  // Phase 1: size every register on our side. No consent is touched here, so
+  // a company with nothing left costs one count and no token refresh.
+  const registers: Register[] = []
+  const sizing = await ctx.forEach('consent', usable, async (consent) => {
+    const candidates = await countRowlessInvoices(supabase, consent.company_id)
+    if (candidates > 0) registers.push({ consent, candidates })
+  })
+  registers.sort((a, b) => a.candidates - b.candidates)
+
+  if (registers.length > 0) {
+    ctx.log.info('registers with row-less invoices, smallest first', {
+      registers: registers.map((r) => ({
+        companyId: r.consent.company_id, provider: r.consent.provider, candidates: r.candidates,
+      })),
+    })
+  }
+
+  // Phase 2: complete them smallest first, each within its share of the run.
+  const deferred: { companyId: string; candidates: number }[] = []
  const totals = {
    companies: 0,
    candidates: 0,
@@ -120,10 +161,10 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines'
    failed: 0,
  }

-  const summary = await ctx.forEach('consent', consents, async (consent, itemCtx) => {
+  const completing = await ctx.forEach('register', registers, async ({ consent, candidates }, itemCtx) => {
    const budgetMs = Math.min(PER_COMPANY_BUDGET_MS, deadline - Date.now())
    if (budgetMs < MIN_COMPANY_BUDGET_MS) {
-      skippedForBudget++
+      deferred.push({ companyId: consent.company_id, candidates })
      return
    }

@@ -134,42 +175,46 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines'
      budgetMs,
    })

-    if (result.candidates > 0) {
-      totals.companies++
-      totals.candidates += result.candidates
-      totals.completed += result.completed
-      totals.headersUpdated += result.headersUpdated
-      totals.remaining += result.remaining
-      totals.notHydrated += result.notHydrated
-      totals.totalMismatch += result.totalMismatch
-      totals.rowsMismatch += result.rowsMismatch
-      totals.failed += result.failed
-      itemCtx.log.info('migrated invoice rows completed for company', {
-        companyId: consent.company_id,
-        provider: consent.provider,
-        candidates: result.candidates,
-        matched: result.matched,
-        completed: result.completed,
-        remaining: result.remaining,
-        notHydrated: result.notHydrated,
-        totalMismatch: result.totalMismatch,
-        rowsMismatch: result.rowsMismatch,
-        hydration: result.hydration,
-      })
-    }
+    // The pass re-reads our side before it touches the consent; a register
+    // the wizard finished between the count and now is neither worked on nor
+    // counted as a company.
+    if (result.candidates === 0) return
+    totals.companies++
+    totals.candidates += result.candidates
+    totals.completed += result.completed
+    totals.headersUpdated += result.headersUpdated
+    totals.remaining += result.remaining
+    totals.notHydrated += result.notHydrated
+    totals.totalMismatch += result.totalMismatch
+    totals.rowsMismatch += result.rowsMismatch
+    totals.failed += result.failed
+    itemCtx.log.info('migrated invoice rows completed for company', {
+      companyId: consent.company_id,
+      provider: consent.provider,
+      candidates: result.candidates,
+      matched: result.matched,
+      completed: result.completed,
+      remaining: result.remaining,
+      notHydrated: result.notHydrated,
+      totalMismatch: result.totalMismatch,
+      rowsMismatch: result.rowsMismatch,
+      hydration: result.hydration,
+    })
  })

-  ctx.log.info('complete-invoice-lines run finished', {
-    ...totals, skippedForBudget, consents: summary.total, consentsStale: scanned.length - consents.length,
-  })
+  if (deferred.length > 0) {
+    ctx.log.warn('run deadline reached; the largest registers wait for the next run', { deferred })
+  }

-  return NextResponse.json({
-    data: {
-      consents: summary.total,
-      consentsStale: scanned.length - consents.length,
-      consentsFailed: summary.failed,
-      skippedForBudget,
-      ...totals,
-    },
-  })
+  const summary = {
+    consents: usable.length,
+    consentsStale: scanned.length - usable.length,
+    consentsFailed: sizing.failed + completing.failed,
+    skippedForBudget: deferred.length,
+    deferred,
+    ...totals,
+  }
+  ctx.log.info('complete-invoice-lines run finished', summary)
+
+  return NextResponse.json({ data: summary })
 })
@@ -0,0 +1,74 @@
+import { describe, it, expect, vi } from 'vitest'
+import type { SupabaseClient } from '@supabase/supabase-js'
+
+/**
+ * The count the cron sizes and orders its run by. It must be the pass's own
+ * predicate (non-draft sales invoices of this company with no rows) evaluated
+ * in the database as a HEAD count, so a company with thousands of complete
+ * invoices costs one indexed query and no rows. The grammar itself
+ * (`invoice_items=is.null` on a to-many embed) is proven against a real
+ * PostgREST in complete-invoice-lines-count.tool.test.ts; this file pins the
+ * query shape and the error contract.
+ */
+
+vi.mock('@/lib/providers/resolve-consent', () => ({ resolveConsent: vi.fn() }))
+vi.mock('@/lib/providers/provider-data-fetcher', () => ({
+  fetchSalesInvoicesDirect: vi.fn(),
+  hydrateSalesInvoices: vi.fn(),
+}))
+vi.mock('@/lib/supabase/fetch-all', () => ({ fetchAllRows: vi.fn() }))
+
+import { countRowlessInvoices } from '../complete-invoice-lines'
+
+interface Call { method: string; args: unknown[] }
+
+/** Thenable query-builder stand-in that records the chain and answers with `response`. */
+function makeSupabase(response: { count: number | null; error: { message: string } | null }) {
+  const calls: Call[] = []
+  const tables: string[] = []
+  const builder: Record<string, unknown> = {}
+  for (const method of ['select', 'eq', 'neq', 'is']) {
+    builder[method] = (...args: unknown[]) => {
+      calls.push({ method, args })
+      return builder
+    }
+  }
+  builder.then = (resolve: (v: unknown) => void, reject: (e: unknown) => void) =>
+    Promise.resolve(response).then(resolve, reject)
+  const from = vi.fn((table: string) => {
+    tables.push(table)
+    return builder
+  })
+  return { supabase: { from } as unknown as SupabaseClient, calls, tables }
+}
+
+describe('countRowlessInvoices', () => {
+  it('counts non-draft sales invoices with no rows in the database, loading none of them', async () => {
+    const { supabase, calls, tables } = makeSupabase({ count: 155, error: null })
+
+    await expect(countRowlessInvoices(supabase, 'co-1')).resolves.toBe(155)
+
+    expect(tables).toEqual(['invoices'])
+    expect(calls).toEqual([
+      { method: 'select', args: ['id, invoice_items(id)', { count: 'exact', head: true }] },
+      { method: 'eq', args: ['company_id', 'co-1'] },
+      { method: 'eq', args: ['document_type', 'invoice'] },
+      { method: 'neq', args: ['status', 'draft'] },
+      { method: 'is', args: ['invoice_items', null] },
+    ])
+  })
+
+  it('throws on a failed count rather than reporting the company as done', async () => {
+    const { supabase } = makeSupabase({ count: null, error: { message: 'canceling statement due to statement timeout' } })
+
+    await expect(countRowlessInvoices(supabase, 'co-1')).rejects.toThrow(
+      'invoices count failed: canceling statement due to statement timeout',
+    )
+  })
+
+  it('reads a missing count as nothing to do', async () => {
+    const { supabase } = makeSupabase({ count: null, error: null })
+
+    await expect(countRowlessInvoices(supabase, 'co-1')).resolves.toBe(0)
+  })
+})
@@ -0,0 +1,94 @@
+import { randomUUID } from 'node:crypto'
+import { describe, it, expect, beforeAll } from 'vitest'
+import { getPool } from '@/tests/pg/setup'
+import { seedCompany } from '@/tests/pg/fixtures'
+import { createToolPgClient } from '@/tests/tool-pg/client'
+import { countRowlessInvoices } from '../complete-invoice-lines'
+
+/**
+ * `invoice_items=is.null` on a to-many embed is resolved by PostgREST, not by
+ * Postgres and not by the type system, and a HEAD count on top of it is one
+ * more thing a mocked client answers whatever the grammar means. The cron
+ * skips every register this count reports as empty, so a count that came
+ * back 0 for everyone would switch the whole cron off, and a count that
+ * ignored the embed filter would size registers by their whole history. Both
+ * pass unit tests. This asserts on the real thing that the count is the
+ * anti-join the pass's own `loadCandidates` evaluates client-side.
+ */
+
+async function insertInvoice(input: {
+  userId: string
+  companyId: string
+  status?: string
+  documentType?: string
+  rows?: number
+}): Promise<string> {
+  const id = randomUUID()
+  await getPool().query(
+    `INSERT INTO public.invoices
+       (id, user_id, company_id, invoice_number, invoice_date, due_date, status, document_type, total)
+     VALUES ($1, $2, $3, $4, '2026-03-14', '2026-04-13', $5, $6, 1250)`,
+    [id, input.userId, input.companyId, `F-${randomUUID()}`, input.status ?? 'sent', input.documentType ?? 'invoice'],
+  )
+  for (let i = 0; i < (input.rows ?? 0); i++) {
+    await getPool().query(
+      `INSERT INTO public.invoice_items (invoice_id, sort_order, description, quantity, unit_price, line_total)
+       VALUES ($1, $2, 'Konsulttid', 1, 1000, 1000)`,
+      [id, i + 1],
+    )
+  }
+  return id
+}
+
+describe('countRowlessInvoices against a real PostgREST', () => {
+  let client: ReturnType<typeof createToolPgClient>
+  let companyId: string
+  let otherCompanyId: string
+
+  beforeAll(async () => {
+    client = createToolPgClient()
+    const a = await seedCompany()
+    const b = await seedCompany()
+    companyId = a.companyId
+    otherCompanyId = b.companyId
+
+    // Three the pass would complete...
+    await insertInvoice({ userId: a.userId, companyId })
+    await insertInvoice({ userId: a.userId, companyId, status: 'paid' })
+    await insertInvoice({ userId: a.userId, companyId, status: 'overdue' })
+    // ...and four it would not: rows already there, a draft, a proforma.
+    await insertInvoice({ userId: a.userId, companyId, status: 'paid', rows: 1 })
+    await insertInvoice({ userId: a.userId, companyId, rows: 3 })
+    await insertInvoice({ userId: a.userId, companyId, status: 'draft' })
+    await insertInvoice({ userId: a.userId, companyId, documentType: 'proforma' })
+    // Another company's row-less invoice must not leak into the count.
+    await insertInvoice({ userId: b.userId, companyId: otherCompanyId })
+  }, 30_000)
+
+  it('counts exactly the non-draft sales invoices with no rows, per company', async () => {
+    expect(await countRowlessInvoices(client, companyId)).toBe(3)
+    expect(await countRowlessInvoices(client, otherCompanyId)).toBe(1)
+  })
+
+  it('agrees with the predicate the pass evaluates client-side', async () => {
+    const { data, error } = await client
+      .from('invoices')
+      .select('id, invoice_items(id)')
+      .eq('company_id', companyId)
+      .eq('document_type', 'invoice')
+      .neq('status', 'draft')
+    expect(error).toBeNull()
+    const rows = (data ?? []) as { id: string; invoice_items: { id: string }[] | null }[]
+    const rowless = rows.filter((row) => (row.invoice_items?.length ?? 0) === 0)
+    expect(rows).toHaveLength(5)
+    expect(rowless).toHaveLength(await countRowlessInvoices(client, companyId))
+  })
+
+  it('reports zero for a company whose invoices all have rows', async () => {
+    const c = await seedCompany()
+    await insertInvoice({ userId: c.userId, companyId: c.companyId, rows: 2 })
+    await insertInvoice({ userId: c.userId, companyId: c.companyId, status: 'paid', rows: 1 })
+
+    expect(await countRowlessInvoices(client, c.companyId)).toBe(0)
+  })
+})
@@ -49,6 +49,30 @@ import { mapSalesInvoice } from './entity-mapper'

 const log = createLogger('extensions/arcim-migration/complete-invoice-lines')

+/**
+ * How many invoices in this company the pass would try to complete: the
+ * predicate `loadCandidates` uses (non-draft sales invoices with no rows),
+ * counted in the database instead of loaded. `invoice_items=is.null` is
+ * PostgREST's anti-join on a to-many embed (the parents whose embed is
+ * empty), so a company whose 2 000 invoices are all complete costs one
+ * indexed count rather than two pages of rows. The cron sizes and orders its
+ * run with this before any consent is touched; zero means the pass would
+ * return before its first provider call. Keep the filters in step with
+ * `loadCandidates`: complete-invoice-lines-count.tool.test.ts checks that
+ * the two agree on a real PostgREST.
+ */
+export async function countRowlessInvoices(supabase: SupabaseClient, companyId: string): Promise<number> {
+  const { count, error } = await supabase
+    .from('invoices')
+    .select('id, invoice_items(id)', { count: 'exact', head: true })
+    .eq('company_id', companyId)
+    .eq('document_type', 'invoice')
+    .neq('status', 'draft')
+    .is('invoice_items', null)
+  if (error) throw new Error(`invoices count failed: ${error.message}`)
+  return count ?? 0
+}
+
 export interface CompleteInvoiceLinesOptions {
  supabase: SupabaseClient
  companyId: string
@@ -499,6 +499,33 @@ describe('scanner behaviour (the net catches, and does not over-catch)', () => {
    expect(phantoms(code)).toEqual(['journal_entries.nope'])
  })

+  it('does not accuse an embed-null filter, the anti-join on a to-many embed', () => {
+    // `invoice_items=is.null` (PostgREST: the parents whose embed is empty)
+    // names the embed the select declares, not a column of invoices. The
+    // grammar is proven on a real PostgREST by
+    // extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines-count.tool.test.ts.
+    const code = [
+      "supabase.from('invoices')",
+      "  .select('id, invoice_items(id)', { count: 'exact', head: true })",
+      "  .eq('company_id', c)",
+      "  .is('invoice_items', null)",
+      "  .not('invoice_items', 'is', null)",
+      "  .filter('invoice_items', 'is', null)",
+    ].join('\n')
+    expect(phantoms(code)).toEqual([])
+  })
+
+  it('still accuses a bare embed name under any other operator, and an undeclared one under is', () => {
+    // Only `is.null` reaches an embed: `.eq('invoice_items', x)` is a phantom
+    // column PostgREST answers 42703 to, and so is `is` on a name the select
+    // never embedded.
+    const code = [
+      "supabase.from('invoices').select('id, invoice_items(id)').eq('invoice_items', 1)",
+      "supabase.from('invoices').select('id').is('invoice_items', null)",
+    ].join('\n')
+    expect(phantoms(code)).toEqual(['invoices.invoice_items', 'invoices.invoice_items'])
+  })
+
  it('does not accuse embedded resource names, aliases or casts', () => {
    const code = [
      "supabase.from('invoices')",
@@ -1457,6 +1457,20 @@ function processChain(
        }
        continue
      }
+      // PostgREST's null filter on an embedded resource (`?invoice_items=is.null`,
+      // the anti-join on a to-many embed: the parents whose embed is empty)
+      // names the embed declared in this chain's select, not a column. The
+      // embed itself was resolved when the aliases were registered, so there
+      // is no column left to check. Only `is` reaches an embed this way:
+      // `.eq('invoice_items', x)` is a real phantom and stays one. Grammar
+      // proven on a real PostgREST by complete-invoice-lines-count.tool.test.ts.
+      const operator = method === 'is' ? 'is' : literalText(args[1])
+      const embedNullFilter =
+        !raw.includes('.') &&
+        aliases.has(raw) &&
+        operator === 'is' &&
+        (method === 'is' || method === 'not' || method === 'filter')
+      if (embedNullFilter) continue
      const resolved = resolvePath(raw, node, 'filter')
      if (!resolved) continue
      push(resolved.table, resolved.column, 'filter', node, raw)