forked from
nekomimi.pet/wisp.place-monorepo
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
concurrent sliding-window backfill, run firehose during backfill, fix revalidate skipping cache invalidation
+17
-5
+14
-2
apps/firehose-service/src/index.ts
+14
-2
apps/firehose-service/src/index.ts
···
97
97
let sitesSynced = 0
98
98
let sitesFailed = 0
99
99
100
-
for (const did of dids) {
100
+
const concurrency = config.backfillConcurrency
101
+
102
+
const processDid = async (did: string) => {
101
103
try {
102
104
const records = await listSiteRecordsForDid(did)
103
105
for (const row of records) {
···
112
114
}
113
115
}
114
116
didsProcessed++
115
-
logger.info(`[Backfill:sites] Progress ${didsProcessed + didsFailed}/${dids.length} DIDs`)
116
117
} catch (err) {
117
118
logger.error(`[Backfill:sites] Failed to list records for DID ${did}`, err)
118
119
didsFailed++
119
120
}
121
+
logger.info(`[Backfill:sites] Progress ${didsProcessed + didsFailed}/${dids.length} DIDs (${sitesSynced} sites synced, ${sitesFailed} sites failed)`)
120
122
}
123
+
124
+
const inFlight = new Set<Promise<void>>()
125
+
for (const did of dids) {
126
+
const task = processDid(did).then(() => { inFlight.delete(task) })
127
+
inFlight.add(task)
128
+
if (inFlight.size >= concurrency) {
129
+
await Promise.race(inFlight)
130
+
}
131
+
}
132
+
await Promise.all(inFlight)
121
133
122
134
logger.info(
123
135
`Phase 2/3 complete: ${didsProcessed} DIDs processed, ${didsFailed} DIDs failed, ${sitesSynced} sites synced, ${sitesFailed} sites failed`,
+2
-2
apps/firehose-service/src/lib/cache-writer.ts
+2
-2
apps/firehose-service/src/lib/cache-writer.ts
···
830
830
})
831
831
}
832
832
833
-
// Notify hosting-service to invalidate its local caches
834
-
// (skip for revalidate/backfill since hosting-service already has the files locally)
833
+
// Notify hosting-service to invalidate its local caches (including negative 404 cache)
834
+
// (skip for backfill since it runs before the hosting-service serves traffic)
835
835
if (!options?.skipInvalidation) {
836
836
await publishCacheInvalidation(did, rkey, 'update')
837
837
}