A Bluesky labeler that labels accounts hosted on PDSes operated by entities other than Bluesky PBC
Add failure backoff for background re-crawling
+84
-27
+1
index.ts
+1
index.ts
···
14
14
host: process.env.SERVER_HOST,
15
15
reconsiderActivePDSForRecrawlingAfterMilliseconds: parseInt(process.env.RECRAWL_ACTIVE_PDS_AFTER_MS ?? "14400000" /* 4h */),
16
16
recrawlKnownPDSAfterNotCrawledForMilliseconds: parseInt(process.env.RECRAWL_KNOWN_PDS_AFTER_MS ?? "43200000" /* 12h */),
17
+
retryFailedCrawlsAfterMinMilliseconds: parseInt(process.env.RETRY_FAILED_CRAWLS_AFTER_MS ?? "43200000" /* 12h */),
17
18
updateLabelDefinitionsEveryMilliseconds: parseInt(process.env.UPDATE_LABEL_DEFINITIONS_EVERY_MS ?? "7200000" /* 2h */),
18
19
allowIssuingLabels: process.env.ACTUALLY_EMIT_LABELS ? process.env.ACTUALLY_EMIT_LABELS === "true" : throwMissingEnvVarError(),
19
20
maxPDSDedicatedLabels: parseInt(process.env.MAX_PDS_DEDICATED_LABELS ?? "30"),
+57
-14
knownPDSStorage.ts
+57
-14
knownPDSStorage.ts
···
30
30
PRIMARY KEY (uri)
31
31
);
32
32
`);
33
+
await this.db.execute(`
34
+
CREATE TABLE IF NOT EXISTS pds_crawling_failures (
35
+
uri TEXT NOT NULL,
36
+
last_attempt DATETIME NOT NULL,
37
+
PRIMARY KEY (uri)
38
+
);
39
+
`);
33
40
}
34
41
35
42
private rowToKnownPDS(row: any): KnownPDS {
···
41
48
return v
42
49
}
43
50
44
-
async* getKnownPDSs(lastCrawledBefore: Date = new Date()) {
51
+
async* getKnownPDSs(lastCrawledBefore: Date = new Date(), lastFailureBefore: Date = new Date()) {
45
52
await this.dbInitLock;
46
53
47
54
let cursor = "";
···
49
56
const result = await this.db.execute({
50
57
sql: `
51
58
SELECT
52
-
uri, last_crawled, total_repos
53
-
FROM known_pds
54
-
WHERE uri > ? AND last_crawled < ?
59
+
k.uri, k.last_crawled, k.total_repos
60
+
FROM known_pds k
61
+
WHERE
62
+
k.uri > ? AND
63
+
k.last_crawled < ? AND
64
+
k.uri NOT IN (
65
+
SELECT f.uri
66
+
FROM pds_crawling_failures f
67
+
WHERE f.last_attempt >= ?
68
+
)
55
69
ORDER BY uri
56
70
LIMIT 100
57
71
`,
58
-
args: [cursor, lastCrawledBefore.toISOString()],
72
+
args: [cursor, lastCrawledBefore.toISOString(), lastFailureBefore.toISOString()],
59
73
});
60
74
61
75
if (!result.rows.length) {
···
80
94
FROM known_pds
81
95
ORDER BY total_repos DESC, uri
82
96
LIMIT ?
83
-
`,
97
+
`,
84
98
args: [limit],
85
99
});
86
100
···
99
113
uri, last_crawled, total_repos
100
114
FROM known_pds
101
115
WHERE uri = ?
102
-
`,
116
+
`,
103
117
args: [uri],
104
118
});
105
119
···
113
127
async upsertKnownPDS(knownPDS: KnownPDS) {
114
128
await this.dbInitLock;
115
129
130
+
const transaction = await this.db.transaction("write");
131
+
132
+
try {
133
+
const result = await transaction.execute({
134
+
sql: `
135
+
INSERT INTO known_pds (uri, last_crawled, total_repos)
136
+
VALUES (?, ?, ?)
137
+
ON CONFLICT (uri) DO UPDATE SET
138
+
last_crawled = excluded.last_crawled,
139
+
total_repos = excluded.total_repos
140
+
`,
141
+
args: [knownPDS.uri, knownPDS.lastCrawled.toISOString(), knownPDS.totalRepos],
142
+
});
143
+
144
+
if (!result.rowsAffected) throw new Error("Failed to upsert known PDS");
145
+
146
+
await transaction.execute({
147
+
sql: `
148
+
DELETE FROM pds_crawling_failures
149
+
WHERE uri = ?
150
+
`,
151
+
args: [knownPDS.uri],
152
+
});
153
+
await transaction.commit();
154
+
} finally {
155
+
transaction.close();
156
+
}
157
+
}
158
+
159
+
async markPDSCrawlFailure(uri: string, at: Date) {
116
160
const result = await this.db.execute({
117
161
sql: `
118
-
INSERT INTO known_pds (uri, last_crawled, total_repos)
119
-
VALUES (?, ?, ?)
120
-
ON CONFLICT (uri) DO UPDATE SET
121
-
last_crawled = excluded.last_crawled,
122
-
total_repos = excluded.total_repos
123
-
`,
124
-
args: [knownPDS.uri, knownPDS.lastCrawled.toISOString(), knownPDS.totalRepos],
162
+
INSERT INTO pds_crawling_failures (uri, last_attempt)
163
+
VALUES (?, ?)
164
+
ON CONFLICT (uri) DO UPDATE SET
165
+
last_attempt = excluded.last_attempt
166
+
`,
167
+
args: [uri, at.toISOString()],
125
168
});
126
169
127
170
if (!result.rowsAffected) throw new Error("Failed to upsert known PDS");
+26
-13
labeler.ts
+26
-13
labeler.ts
···
14
14
host: string | undefined,
15
15
reconsiderActivePDSForRecrawlingAfterMilliseconds: number,
16
16
recrawlKnownPDSAfterNotCrawledForMilliseconds: number,
17
+
retryFailedCrawlsAfterMinMilliseconds: number;
17
18
updateLabelDefinitionsEveryMilliseconds: number,
18
19
allowIssuingLabels: boolean,
19
20
maxPDSDedicatedLabels: number,
···
41
42
42
43
private reconsiderActivePDSForRecrawlingAfterMilliseconds: number;
43
44
private recrawlKnownPDSAfterNotCrawledForMilliseconds: number;
45
+
private retryFailedCrawlsAfterMinMilliseconds: number;
44
46
private updateLabelDefinitionsEveryMilliseconds: number;
45
47
private maxPDSDedicatedLabels: number;
46
48
···
62
64
this.knownPDSStorage = new KnownPDSStorage(this.server.db);
63
65
this.crawler = new PDSCrawler(options.maxExpectedReposPerPDS);
64
66
65
-
this.reconsiderActivePDSForRecrawlingAfterMilliseconds = options.reconsiderActivePDSForRecrawlingAfterMilliseconds
66
-
this.recrawlKnownPDSAfterNotCrawledForMilliseconds = options.recrawlKnownPDSAfterNotCrawledForMilliseconds
67
+
this.reconsiderActivePDSForRecrawlingAfterMilliseconds = options.reconsiderActivePDSForRecrawlingAfterMilliseconds;
68
+
this.recrawlKnownPDSAfterNotCrawledForMilliseconds = options.recrawlKnownPDSAfterNotCrawledForMilliseconds;
69
+
this.retryFailedCrawlsAfterMinMilliseconds = options.retryFailedCrawlsAfterMinMilliseconds;
67
70
this.updateLabelDefinitionsEveryMilliseconds = options.updateLabelDefinitionsEveryMilliseconds;
68
71
69
72
this.allowIssuingLabels = options.allowIssuingLabels;
···
115
118
const correctLabels = this.labelDefiner.determinePDSLabels(pds);
116
119
const correctLabelIDs = correctLabels.map((l) => l.identifier);
117
120
let totalRepos = 0;
118
-
for await (const repo of this.crawler.getPDSRepos(pds)) {
119
-
await this.processDIDWithLabels(repo.did, correctLabelIDs);
120
-
totalRepos++;
121
+
122
+
try {
123
+
for await (const repo of this.crawler.getPDSRepos(pds)) {
124
+
await this.processDIDWithLabels(repo.did, correctLabelIDs);
125
+
totalRepos++;
126
+
}
127
+
128
+
await this.knownPDSStorage.upsertKnownPDS({
129
+
uri: pds,
130
+
lastCrawled: new Date(),
131
+
totalRepos,
132
+
});
133
+
console.log(`Done crawling ${pds} with a total of ${totalRepos} repos`);
134
+
} catch {
135
+
await this.knownPDSStorage.markPDSCrawlFailure(pds, new Date());
136
+
console.log(`Failed to crawl ${pds}, took note`);
121
137
}
122
-
123
-
await this.knownPDSStorage.upsertKnownPDS({
124
-
uri: pds,
125
-
lastCrawled: new Date(),
126
-
totalRepos,
127
-
});
128
-
console.log(`Done crawling ${pds} with a total of ${totalRepos} repos`);
129
138
}
130
139
131
140
private async processSingleDID(pds: string, did: `did:${string}:${string}`) {
···
195
204
}
196
205
197
206
private async longevityBasedProcessing() {
198
-
for await (const knownPDS of this.knownPDSStorage.getKnownPDSs(new Date(new Date().getTime() - this.recrawlKnownPDSAfterNotCrawledForMilliseconds))) {
207
+
const nowMS = new Date().getTime();
208
+
const generator = this.knownPDSStorage.getKnownPDSs(
209
+
new Date(nowMS - this.recrawlKnownPDSAfterNotCrawledForMilliseconds),
210
+
new Date(nowMS - this.retryFailedCrawlsAfterMinMilliseconds));
211
+
for await (const knownPDS of generator) {
199
212
console.log(`Scheduling known PDS ${knownPDS.uri} for re-crawling`);
200
213
this.scheduleTask(async () => {
201
214
console.log(`Re-crawling known PDS ${knownPDS.uri}`);