Merge branch 'rebuild' into voyager-3

app/services/group_dates.ts

@@ -90,33 +90,49 @@ export function sampleCount(
  * server's span.
  */
 export async function sampleGroup(
-  client: XoverClient,
+  use: <T>(work: (client: XoverClient) => Promise<T>) => Promise<T>,
   name: string,
-  options: { every: number; size: number },
+  options: { every: number; size: number; maxFailures?: number },
   now = new Date()
-): Promise<{ samples: DateSample[]; low: number; high: number; commands: number }> {
-  const info = await client.group(name)
+): Promise<{
+  samples: DateSample[]
+  low: number
+  high: number
+  commands: number
+  failed: string[]
+}> {
+  const info = await use((client) => client.group(name))
   const samples: DateSample[] = []
+  const failed: string[] = []
   let commands = 1
+  // Each point on a connection of its own (GROUP again): a reply that times out (Eweka stalls some requests for
+  // minutes; 16 groups never got through on one connection) costs that point, not the group.
   for (const point of samplePoints(
     info.low,
     info.high,
     sampleCount(info.low, info.high, options.every),
     options.size
   )) {
-    commands++
-    let date = medianDate(
-      await client.xover(point, Math.min(point + options.size - 1, info.high)),
-      now
-    )
-    if (!date) {
-      commands++
-      date = medianDate(
-        await client.xover(point, Math.min(point + options.size * 32 - 1, info.high)),
-        now
-      )
+    if (failed.length >= (options.maxFailures ?? 3)) break
+    try {
+      const date = await use(async (client) => {
+        await client.group(name)
+        commands += 2
+        const slice = medianDate(
+          await client.xover(point, Math.min(point + options.size - 1, info.high)),
+          now
+        )
+        if (slice) return slice
+        commands++
+        return medianDate(
+          await client.xover(point, Math.min(point + options.size * 32 - 1, info.high)),
+          now
+        )
+      })
+      if (date) samples.push({ article: point, postedAt: date })
+    } catch (error: any) {
+      failed.push(`${point}: ${error.message}`)
     }
-    if (date) samples.push({ article: point, postedAt: date })
   }
-  return { samples: monotone(samples), low: info.low, high: info.high, commands }
+  return { samples: monotone(samples), low: info.low, high: info.high, commands, failed }
 }

commands/ranges_dates.ts

@@ -48,10 +48,10 @@ export default class RangesDates extends BaseCommand {
     }
   }
 
-  private groups() {
+  private groups(missing = this.missing) {
     const q = db.from('groups').where('active', true).select('id', 'name').orderBy('priority')
     if (this.group) q.where('name', 'like', this.group)
-    if (this.missing)
+    if (missing)
       q.whereNotExists((sub) =>
         sub.from('group_dates').whereRaw('group_dates.group_id = groups.id')
       )
@@ -85,9 +85,11 @@ export default class RangesDates extends BaseCommand {
     const worker = async () => {
       for (let g = queue.shift(); g; g = queue.shift()) {
         try {
-          const result = await NntpPool.use((client) =>
-            sampleGroup(client, g.name, { every, size: 32 })
-          )
+          const result = await sampleGroup((work) => NntpPool.use(work), g.name, {
+            every,
+            size: 32,
+          })
+          for (const f of result.failed) this.logger.warning(`${g.name} at ${f}`)
           await db.transaction(async (trx) => {
             await trx.from('group_dates').where('group_id', g.id).delete()
             if (result.samples.length > 0)
@@ -129,7 +131,7 @@ export default class RangesDates extends BaseCommand {
 
   /** Every range of the groups sampled gets est_date from the group's curve (1000 a statement). */
   private async estimate() {
-    const groups = await this.groups().whereExists((sub) =>
+    const groups = await this.groups(false).whereExists((sub) =>
       sub.from('group_dates').whereRaw('group_dates.group_id = groups.id')
     )
     let updated = 0