Reliability and performance improvements

labeler.ts

···
     private authenticatedRPC: Client | undefined;
 
     private pdsProcessingMutex = new NamedMutex();
+    private alreadyScheduledForCrawling = new Set<string>();
 
     private initialized: Promise<void>;
 
···
 
         const shouldHaveLabels = this.pdsFilter(pds);
 
-        if (typeof maybeKnownPDS === "undefined" && shouldHaveLabels) {
+        if (typeof maybeKnownPDS === "undefined" && shouldHaveLabels && !this.alreadyScheduledForCrawling.has(pds)) {
             this.taskProcessor.schedule(async () => {
                 await this.pdsProcessingMutex.runExclusive(pds, async () => {
                     // this is a task that's scheduled async, must confirm it is still unknown within the mutex
···
                         console.log(`Crawling new PDS discovered when processing labels for ${becauseOfDID}: ${pds}`);
                         await this.processPDS(pds);
                     }
+                    this.alreadyScheduledForCrawling.delete(pds);
                 });
-            }, 15*60*1000 /* 15 minutes */);
+            }, 15 * 60 * 1000 /* 15 minutes */).catch((e) => {
+                console.log(`Error processing PDS ${pds}: ${e}`);
+            });
+            this.alreadyScheduledForCrawling.add(pds);
         }
     }
 
···
             stripWWW: false,
             defaultProtocol: "https",
         }
+
+        const maxConcurrent = 10;
+        const pendingDIDs: Promise<void>[] = [];
 
         try {
             for await (const repo of this.crawler.getPDSRepos(pds)) {
-                try {
-                    const realPDSOfRepo = await this.processDID(repo.did, false, false, true);
-                    if (normalizeUrl(realPDSOfRepo, normOptions) === normalizeUrl(pds, normOptions)) {
-                        totalRepos++;
+                const did = repo.did;
+                const promise = (async () => {
+                    try {
+                        const realPDSOfRepo = await this.processDID(did, false, false, true);
+                        if (normalizeUrl(realPDSOfRepo, normOptions) === normalizeUrl(pds, normOptions)) {
+                            totalRepos++;
+                            if(totalRepos % 1000 === 0) {
+                                console.log(`Crawling ${pds}: processed ${totalRepos} repos so far`);
+                            }
+                        }
+                    } catch (e) {
+                        console.log(`Failed to process ${did} within PDS ${pds}: ${e}`);
                     }
-                } catch (e) {
-                    console.log(`Failed to process ${repo} within PDS ${pds}: ${e}`);
+                })();
+                pendingDIDs.push(promise);
+
+                // Process in batches of maxConcurrent to limit memory usage
+                if (pendingDIDs.length >= maxConcurrent) {
+                    await Promise.all(pendingDIDs);
+                    pendingDIDs.length = 0;
                 }
+            }
+
+            // Wait for any remaining DIDs to be processed
+            if (pendingDIDs.length > 0) {
+                await Promise.all(pendingDIDs);
             }
 
             await this.knownPDSStorage.upsertKnownPDS({
···
                 totalRepos,
             });
             console.log(`Done crawling ${pds} with a total of ${totalRepos} repos`);
-        } catch(e) {
+        } catch (e) {
             await this.knownPDSStorage.markPDSCrawlFailure(pds, new Date());
             console.log(`Failed to crawl ${pds}, took note: ${e}`);
         }
+        this.alreadyScheduledForCrawling.delete(pds);
     }
 
     private async processDID(did: DID, considerAddingToLists: boolean, scheduleUnknownPDSForCrawling: boolean, mayUseCache: boolean): Promise<string> {
···
                         continue;
                     }
 
-                    this.taskProcessor.schedule(async () => {
-                        await this.processEvent(event.kind, event.did, event.time_us, clockSkew, callback);
-                    }, 10*1000 /* 10 seconds */, 0); // schedule with highest priority
+                    this.taskProcessor.schedule(
+                        async () => {
+                            await this.processEvent(event.kind, event.did, event.time_us, clockSkew, callback);
+                        },
+                        10 * 1000, // 10 seconds
+                        0 // schedule with highest priority
+                    ).catch((e) => {
+                        console.log(`Failed to process event ${event.kind} for ${event.did}: ${e}`);
+                    });
                 }
             } catch (e) {
                 console.log(`Failed to consume jetstream: ${e}`);
···
             const maybeKnownPDS = await this.knownPDSStorage.getKnownPDS(pds);
 
             if (typeof maybeKnownPDS === "undefined") {
+                if (this.alreadyScheduledForCrawling.has(pds)) {
+                    return;
+                }
+                this.alreadyScheduledForCrawling.add(pds);
                 console.log("Crawling previously undiscovered independent PDS: " + pds);
                 await this.processPDS(pds);
             } else if (Date.now() - maybeKnownPDS.lastCrawled.getTime() > this.reconsiderActivePDSForRecrawlingAfterMilliseconds) {
+                if (this.alreadyScheduledForCrawling.has(pds)) {
+                    return;
+                }
+                this.alreadyScheduledForCrawling.add(pds);
                 console.log("Re-crawling already known independent PDS based on detected activity: " + pds);
                 await this.processPDS(pds);
             } else {
···
                 return;
             }
 
-            this.taskProcessor.schedule(this.refreshDIDOrCrawlPDS.bind(this, pds, did), 15*60*1000 /* 15 minutes */);
+            this.taskProcessor.schedule(this.refreshDIDOrCrawlPDS.bind(this, pds, did), 15 * 60 * 1000 /* 15 minutes */).catch(
+                (e) => console.log(`Failed to process ${did} after detected activity: ${e}`)
+            );
         });
     }
 
···
             new Date(nowMS - this.recrawlKnownPDSAfterNotCrawledForMilliseconds),
             new Date(nowMS - this.retryFailedCrawlsAfterMinMilliseconds));
         for await (const knownPDS of generator) {
+            if (this.alreadyScheduledForCrawling.has(knownPDS.uri)) {
+                continue;
+            }
             console.log(`Scheduling known PDS ${knownPDS.uri} for re-crawling`);
             this.taskProcessor.schedule(async () => {
                 await this.pdsProcessingMutex.runExclusive(knownPDS.uri, async () => {
···
                     } else {
                         console.log(`Removing known PDS ${knownPDS.uri}`);
                         await this.knownPDSStorage.removeKnownPDS(knownPDS.uri);
+                        this.alreadyScheduledForCrawling.delete(knownPDS.uri);
                     }
                 });
-            }, 15*60*1000 /* 15 minutes */);
+            }, 15 * 60 * 1000 /* 15 minutes */).catch((e) => {
+                console.log(`Failed to reprocess PDS ${knownPDS.uri}: ${e}`);
+            });
+            this.alreadyScheduledForCrawling.add(knownPDS.uri);
         }
     }
 
···
                 if (++totalAttempted === totalToProcess) {
                     resolveAllAttempted();
                 }
-            }, 15*1000 /* 15 seconds */);
+            }, 15 * 1000 /* 15 seconds */, 200).catch(
+                (e) => console.log(`Failed to reprocess labels of ${uri}: ${e}`)
+            );
             // avoid having too many in-flight tasks at once,
             // by waiting until the queue is unsaturated (25% of the workers are free) to schedule more
             // otherwise we'd load dozens of thousands of tasks into memory
···
         wrapAsyncInCatch(this.existingLabelsProcessing());
 
         setInterval(() => {
-            this.taskProcessor.schedule(this.longevityBasedProcessing.bind(this), 15*60*1000 /* 15 minutes */);
+            this.taskProcessor.schedule(this.longevityBasedProcessing.bind(this), 15 * 60 * 1000 /* 15 minutes */);
         }, 60 * 1000);
 
         setInterval(() => {
-            this.taskProcessor.schedule(this.updateLabelerLabels.bind(this, false), 1*60*1000 /* 1 minute */);
+            this.taskProcessor.schedule(this.updateLabelerLabels.bind(this, false), 1 * 60 * 1000 /* 1 minute */);
         }, this.updateLabelDefinitionsEveryMilliseconds);
 
         setInterval(() => {