/* * **놓친 구간을 backfill 해 채운다** — 밀어 받는 길의 짝. * * ── 왜 필요한가 (2026-08-31, 인티그레이션 레인 지적) ────────────────────── * 훅은 반드시 놓친다 — 우리가 내려가 있을 때, 그쪽이 못 보냈을 때, 네트워크가 끊겼을 때. 받는 쪽이 * 번호로 그것을 **알아채게** 되었지만(§`takeInSequence`), 알아챈 뒤 **그 사이를 채우는 길이 없었다.** * 설계가 그 길을 적어 두었는데 만들지 않았다. * * ── 같은 검사를 지난다 ─────────────────────────────────────────────────── * backfill 한 것도 `takeInSequence` 를 지난다. 여기서 따로 만들면 두 길이 다른 규칙으로 받고, 한쪽만 * 고쳐지는 날 어긋난 그 길에서만 사실이 사라진다. * * ── 멈추는 조건을 셋 둔다 ──────────────────────────────────────────────── * 되풀이해서 요청하는 경로는 **끝나지 않는 것**이 가장 흔한 결함이다. * * 빈 답 더 줄 것이 없다. 끝 * 번호가 안 나아간다 같은 것을 다시 받았다 — 무한히 요청하는 것을 막는다 * 구멍 원본의 아웃박스에 구멍이 있다. 이어 받으면 커서가 그 번호를 지나쳐 적힌다 * * 그리고 **쪽 상한**을 둔다. 상한에 닿았으면 조용히 끝내지 않고 말한다 — 조용히 자르면 사람은 다 받은 * 줄로 읽는다. */ import { seqOf, takeInSequence, withSeq } from './hook-contract.js' import type { InboundBatch } from './reference-adapter.js' import { twinWarn } from '../../engine/log.js' /** 한 번 채우고 나서 남는 사실. */ export interface FillOutcome { /** 트윈에 넣은 레코드 수. */ applied: number /** * ingest 가 낸 네 갈래의 합 — **두 갈래로는 셈이 안 닫힌다.** * * ``` * 두 갈래 157 − 46 − 2 = 109 「설명 못 함」 진실은 되풀이 108 + 어휘 1 * 네 갈래 157 − 46 − 2 − 108 − 1 = 0 * ``` * * 이 값이 없으면 원본 쪽에서 「보낸 것 20개 중 event 가 9개인데 나머지는 어디 갔나」를 셀 수 * 없다. 실제로 인티그레이션 레인이 그것을 물었고, 그 수가 트윈에 저장되지 않아 답을 못 했다. */ duplicates: number rejected: number /** * 왜 거절됐나 — **사유별 수.** 수만 내면 「26건이 안 들어온다」에서 멈춘다. * * 목록 전체가 아니라 사유별로 센다. 같은 이유로 거절된 26건이 26줄이면 읽는 사람이 그 줄을 * 세어야 하고, 다른 이유 하나가 그 안에 묻힌다. */ rejectedReasons?: Record vocabulary: number /** 시각을 못 읽어 ingest 시각으로 떨어진 것 — 종류별. 비어 있으면 전부 자기 시각을 말했다. */ undated?: Record /** 몇 쪽을 받았나. */ pages: number /** 커서가 어디까지 갔나 — 아무것도 못 받았으면 없다. */ lastSeq?: number /** * 이 실행에서 **가장 낮게 받아들인 번호** — 밑에서부터 다시 받았을 때 이것이 곧 「우리가 가진 * 가장 낮은 번호」다. */ lowestSeq?: number /** 왜 멈췄나 — `done` · `gap` · `no-progress` · `page-limit` · `refused`. */ stopped: 'done' | 'gap' | 'no-progress' | 'page-limit' | 'refused' /** 멈춘 이유의 사람 말. 알리지 않고 끝내지 않는다. */ detail?: string /** * 마지막으로 받은 묶음이 말한 **원본의 번호 범위** — 원본이 안 말하면 없다. * * 우리 커서와 견주는 자리가 이 값을 쓴다. 없으면 「맞다」가 아니라 **「모른다」**로 답해야 한다. */ sourceRange?: { first?: number; last?: number } } /** 몇 쪽까지 받나 — 한 번의 채우기가 서버를 오래 붙잡지 않게. */ const MAX_PAGES = 50 /** * 그 단위의 놓친 구간을 채운다. * * `ingest` 를 부르는 쪽이 넘겨 준다 — 훅과 같은 이유다(이 파일이 엔진을 직접 부르면 시험이 엔진을 * 세워야 한다). */ export async function fillSince(args: { cfg: Record site: { siteId: string } /** 번호를 매기는 단위 — 커서 열쇠의 절반. 커넥터가 만들지 않는다. */ scope: string /** 커서에 적힌 열쇠 — `pushCursorKey(instanceId, scope)`. */ cursorKey: string /** 지금 커서 — 이 함수가 읽고, 나아간 만큼 되돌려 준다. */ cursor: unknown fetch: (cfg: Record, site: { siteId: string }, scope: string, since: number) => Promise ingest: (records: unknown[]) => { applied: number; duplicates: number; rejected: unknown[] } /** * 넣기 **직전에** 부른다 — 이 쪽의 레코드를 넘긴다. * * 중복 판정 기억을 저널에서 채우는 자리다(§`fill-dedupe.ts`). 그 기억은 프로세스 안에만 있어서 * 재기동하면 비워지고, 그러면 이미 저널에 있는 사실도 「처음 본 것」이 된다. * * `ingest` 는 동기라 여기서 채울 수 없다. 이 훅은 비동기이므로 저널을 읽을 수 있다. * * 실패해도 넣기를 멈추지 않는다 — 기억을 못 채운 것이 사실을 못 받을 이유는 아니다. */ beforeIngest?: (records: unknown[]) => Promise saveCursor: (cursor: Record) => Promise maxPages?: number }): Promise { const { cfg, site, scope, cursorKey, fetch, ingest, saveCursor , beforeIngest } = args const maxPages = args.maxPages ?? MAX_PAGES let cursor = args.cursor let applied = 0 let duplicates = 0 let rejectedCount = 0 const rejectedReasons: Record = {} let vocabulary = 0 const undated: Record = {} /* 원본이 말한 번호 범위 — 마지막으로 받은 묶음의 것을 들고 나간다. 안 말하면 끝까지 없다. */ let sourceRange: { first?: number; last?: number } | undefined /* 이 실행에서 가장 낮게 받아들인 번호 — 커서에 firstSeq 로 적는다. */ let lowestSeq: number | undefined let pages = 0 for (; pages < maxPages; pages++) { const since = seqOf(cursor, cursorKey) ?? 0 let batch: InboundBatch try { batch = await fetch(cfg, site, scope, since) } catch (e: any) { return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'refused', detail: String(e?.message ?? e) } } if (batch?.sourceRange) sourceRange = batch.sourceRange const items = batch?.items ?? [] /* 빈 답은 더 줄 것이 없다는 뜻이다. 오류가 아니다. */ if (!items.length) return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'done' } const taken = takeInSequence({ items, scope, lastSeq: seqOf(cursor, cursorKey) }) if (taken.refused) { return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'refused', detail: taken.refused } } if (taken.records.length) { if (beforeIngest) { await beforeIngest(taken.records).catch(e => twinWarn(`[twin-fill] 중복 판정 기억을 못 채웠다 — ${String(e?.message ?? e)}. 넣기는 그대로 간다`) ) } const result = ingest(taken.records) applied += result.applied duplicates += result.duplicates ?? 0 rejectedCount += result.rejected?.length ?? 0 for (const r of result.rejected ?? []) { /* 사유가 여럿이면 첫 줄로 묶는다 — 뒤는 대개 같은 원인의 곁가지다. */ const why = String((r as any)?.errors?.[0] ?? '(사유 없음)').slice(0, 120) rejectedReasons[why] = (rejectedReasons[why] ?? 0) + 1 } vocabulary += (result as any).vocabulary ?? 0 for (const [k, n] of Object.entries((result as any).undated ?? {})) undated[k] = (undated[k] ?? 0) + (n as number) } /* * **번호가 나아가야 다음 쪽을 요청한다.** 원본이 빈 묶음에 물어본 번호를 그대로 돌려주는 계약이면, * 나아감을 보지 않고 요청하면 무한히 돈다. */ if (taken.advancedSeq === undefined || taken.advancedSeq <= since) { if (taken.gap) return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages: pages + 1, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'gap', detail: gapWord(taken.gap) } return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages: pages + 1, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'no-progress', detail: `번호가 ${since} 에서 나아가지 않았다 — 같은 것을 다시 받았다` } } if (lowestSeq === undefined) lowestSeq = taken.lowestSeq /* 밑에서부터 다시 받았으면 그 번호가 곧 우리가 가진 가장 낮은 번호다. */ cursor = withSeq(cursor, cursorKey, taken.advancedSeq, taken.firstSeq ?? lowestSeq) await saveCursor(cursor as Record).catch(err => twinWarn(`[twin-fill] 번호를 커서에 적지 못했다 (${cursorKey}=${taken.advancedSeq}) — ${err?.message ?? err}`) ) /* * **구멍이 있으면 멈춘다.** 원본의 아웃박스에 구멍이 있는 것이고, 이어 받으면 커서가 그 번호를 * 지나쳐 적혀 빠진 사실이 다시 오지 않는다. */ if (taken.gap) { return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages: pages + 1, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'gap', detail: gapWord(taken.gap) } } } /* 상한에 닿았다 — 조용히 끝내지 않는다. 사람이 다 받은 줄로 읽으면 안 된다. */ return { applied, duplicates, rejected: rejectedCount, ...(Object.keys(rejectedReasons).length ? { rejectedReasons } : {}), vocabulary, ...(Object.keys(undated).length ? { undated } : {}), pages, ...seqPart(cursor, cursorKey), ...(sourceRange ? { sourceRange } : {}), ...(lowestSeq !== undefined ? { lowestSeq } : {}), stopped: 'page-limit', detail: `${maxPages}쪽까지 받고 멈췄다 — 남은 것이 있을 수 있다. 다음 주기가 이어 받는다` } } function seqPart(cursor: unknown, key: string): { lastSeq?: number } { const seq = seqOf(cursor, key) return seq === undefined ? {} : { lastSeq: seq } } function gapWord(gap: { expectedSeq: number; lastSeq: number }): string { return `번호가 비었다 — ${gap.lastSeq} 다음이 ${gap.expectedSeq} 이어야 한다. 원본의 아웃박스에 구멍이 있다` }