@@ -19,6 +19,8 @@ interface ReconciliationRow {
1919
2020export interface ReconciliationWalk {
2121 connectorId : string
22+ /** Resumes after this document id, the `lastId` an earlier walk stopped at. */
23+ startAfterId ?: string
2224 /**
2325 * Evaluated over the window's rows, which carry only the document's id, connector, exclusion,
2426 * archival, tombstone, seen, ACL and content-hash columns.
@@ -31,6 +33,13 @@ export interface ReconciliationWalk {
3133 onPage : ( rows : ReconciliationRow [ ] ) => Promise < void >
3234}
3335
36+ export interface ReconciliationWalkResult {
37+ /** False when the deadline stopped the walk. */
38+ finished : boolean
39+ /** The last document id the walk covered, from which a stopped walk resumes. */
40+ lastId : string | undefined
41+ }
42+
3443/** A type alias, not an interface, so it satisfies `db.execute`'s row-record constraint. */
3544type WindowScan = {
3645 size : number
@@ -97,29 +106,31 @@ async function scanWindow(walk: ReconciliationWalk, afterId: string | undefined)
97106}
98107
99108/**
100- * Walks a connector's owned documents in id order, one window per statement, handing the rows
101- * matching `condition` to `onPage` in pages of at most `pageSize`. A window without matches still
102- * advances the walk, which ends after the first window shorter than a full one. Returns false when
103- * the deadline stopped it .
109+ * Walks a connector's owned documents in id order from `startAfterId` , one window per statement,
110+ * handing the rows matching `condition` to `onPage` in pages of at most `pageSize`. A window
111+ * without matches still advances the walk, which ends after the first window shorter than a full
112+ * one .
104113 */
105- export async function walkReconciliationWindows ( walk : ReconciliationWalk ) : Promise < boolean > {
106- let after : string | undefined
114+ export async function walkReconciliationWindows (
115+ walk : ReconciliationWalk
116+ ) : Promise < ReconciliationWalkResult > {
117+ let covered = walk . startAfterId
107118 for ( ; ; ) {
108- if ( Date . now ( ) >= walk . deadlineAt ) return false
119+ if ( Date . now ( ) >= walk . deadlineAt ) return { finished : false , lastId : covered }
109120 await walk . beforePage ( )
110- if ( Date . now ( ) >= walk . deadlineAt ) return false
111- const { size, last, ids, tombstoned } = await scanWindow ( walk , after )
121+ if ( Date . now ( ) >= walk . deadlineAt ) return { finished : false , lastId : covered }
122+ const { size, last, ids, tombstoned } = await scanWindow ( walk , covered )
112123 for ( let offset = 0 ; offset < ids . length ; offset += walk . pageSize ) {
113124 if ( offset > 0 ) await walk . beforePage ( )
114125 /** Materialized ids are acted on only inside the budget, so a late page is left for the next run. */
115- if ( Date . now ( ) >= walk . deadlineAt ) return false
116- await walk . onPage (
117- ids
118- . slice ( offset , offset + walk . pageSize )
119- . map ( ( id , index ) => ( { id, tombstoned : tombstoned [ offset + index ] } ) )
120- )
126+ if ( Date . now ( ) >= walk . deadlineAt ) return { finished : false , lastId : covered }
127+ const page = ids . slice ( offset , offset + walk . pageSize )
128+ await walk . onPage ( page . map ( ( id , index ) => ( { id, tombstoned : tombstoned [ offset + index ] } ) ) )
129+ covered = page . at ( - 1 )
130+ }
131+ if ( size < RECONCILIATION_WINDOW_SIZE || ! last ) {
132+ return { finished : true , lastId : last ?? covered }
121133 }
122- if ( size < RECONCILIATION_WINDOW_SIZE || ! last ) return true
123- after = last
134+ covered = last
124135 }
125136}
0 commit comments