diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 7ceeac1..1938d31 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -1,5 +1,6 @@ import db from '@adonisjs/lucid/services/db' -import type { EdgeBatch } from '#lib/edge/index' +import type { TransactionClientContract } from '@adonisjs/lucid/types/database' +import type { EdgeBatch, ProbeCandidate, ReleaseChunk } from '#lib/edge/index' import ChunkPartitions from '#services/chunk_partitions' export class StaleBatch extends Error {} @@ -23,66 +24,28 @@ class BatchStore { await ChunkPartitions.ensure() return db.transaction(async (trx) => { const range = await trx.from('ranges').where('id', rangeId).forUpdate().first() - if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`) - if (Number(range.next_id) !== batch.from) throw new StaleBatch(`Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`) - const result: StoredBatch = { releases: batch.releases.length, newReleases: 0, candidates: 0, fragments: batch.fragments.length } - - // Releases, sorted by key so concurrent batches lock them in the same order. - const releases = [...batch.releases].sort((a, b) => a.key.localeCompare(b.key)) - const ids = new Map() - for (let i = 0; i < releases.length; i += 1000) { - const chunk = releases.slice(i, i + 1000) - const { rows } = await trx.rawQuery( - `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2) - SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2 - FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean) - ON CONFLICT (release_key) DO UPDATE SET - files = releases.files + excluded.files, - segments = releases.segments + excluded.segments, - bytes = releases.bytes + excluded.bytes, - chunks = releases.chunks + 1, - posted_at = least(releases.posted_at, excluded.posted_at), - file_total = coalesce(releases.file_total, excluded.file_total), - first_article = least(releases.first_article, excluded.first_article), - last_article = greatest(releases.last_article, excluded.last_article), - readable = releases.readable OR excluded.readable, - has_par2 = releases.has_par2 OR excluded.has_par2, - last_seen_at = now() - RETURNING id, release_key, (xmax = 0) AS inserted`, - [ - JSON.stringify( - chunk.map((r) => ({ - key: r.key, name: r.name, source: r.source, poster: r.poster, groups: r.groups, posted_at: r.postedAt, file_total: r.fileTotal, - files: r.files, segments: r.segments, bytes: r.bytes, first_article: r.firstArticle, last_article: r.lastArticle, readable: r.readable, has_par2: r.hasPar2, - })) - ), - ] + if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) + throw new StaleBatch(`Range ${rangeId} is not leased to this node`) + if (Number(range.next_id) !== batch.from) + throw new StaleBatch( + `Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}` ) - for (const row of rows) { - ids.set(row.release_key, Number(row.id)) - if (row.inserted) result.newReleases++ - } + const result: StoredBatch = { + releases: batch.releases.length, + newReleases: 0, + candidates: 0, + fragments: batch.fragments.length, } - // One write-once chunk per release. - for (let i = 0; i < releases.length; i += 500) { - const rows = releases.slice(i, i + 500).map((r) => ({ - release_id: ids.get(r.key)!, - range_id: rangeId, - batch_from: batch.from, - files: r.files, - segments: r.segments, - payload: r.payload, - })) - if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows) - } - - const candidates = batch.candidates.filter((c) => ids.has(c.releaseKey)) - for (let i = 0; i < candidates.length; i += 2000) { - const rows = candidates.slice(i, i + 2000).map((c) => ({ release_id: ids.get(c.releaseKey)!, kind: c.kind, message_id: c.messageId, file: c.file })) - const added = await trx.knexQuery().table('probe_candidates').insert(rows).onConflict(['release_id', 'kind', 'message_id']).ignore().returning('id') - result.candidates += added.length - } + const written = await this.writeReleases( + trx, + rangeId, + batch.from, + batch.releases, + batch.candidates + ) + result.newReleases = written.newReleases + result.candidates = written.candidates if (batch.counts.length > 0) { await trx.rawQuery( @@ -95,7 +58,14 @@ class BatchStore { for (const s of batch.samples) { await trx.insertQuery().table('rule_samples').insert({ - range_id: rangeId, rule: s.rule, rule_key: s.ruleKey, release_key: s.key, name: s.name, files: s.files, segments: s.segments, payload: s.payload, + range_id: rangeId, + rule: s.rule, + rule_key: s.ruleKey, + release_key: s.key, + name: s.name, + files: s.files, + segments: s.segments, + payload: s.payload, }) } @@ -113,10 +83,112 @@ class BatchStore { await trx .from('ranges') .where('id', rangeId) - .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), heartbeat_at: new Date() }) + .update({ + next_id: batch.next, + headers: db.raw('headers + ?', [batch.headers]), + heartbeat_at: new Date(), + }) return result }) } + + /** + * Release aggregates in the catalog, one write-once chunk per release, and + * probe candidates (a batch's, or files the fragment merger completed). + * Releases are sorted by key, so concurrent writers lock them in the same order. + */ + async writeReleases( + trx: TransactionClientContract, + rangeId: number, + batchFrom: number, + releases: ReleaseChunk[], + candidates: ProbeCandidate[] + ) { + const out = { newReleases: 0, candidates: 0, ids: new Map() } + const ids = out.ids + // Releases, sorted by key so concurrent batches lock them in the same order. + const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key)) + for (let i = 0; i < sorted.length; i += 1000) { + const chunk = sorted.slice(i, i + 1000) + const { rows } = await trx.rawQuery( + `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2) + SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2 + FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean) + ON CONFLICT (release_key) DO UPDATE SET + files = releases.files + excluded.files, + segments = releases.segments + excluded.segments, + bytes = releases.bytes + excluded.bytes, + chunks = releases.chunks + 1, + posted_at = least(releases.posted_at, excluded.posted_at), + file_total = coalesce(releases.file_total, excluded.file_total), + first_article = least(releases.first_article, excluded.first_article), + last_article = greatest(releases.last_article, excluded.last_article), + readable = releases.readable OR excluded.readable, + has_par2 = releases.has_par2 OR excluded.has_par2, + last_seen_at = now() + RETURNING id, release_key, (xmax = 0) AS inserted`, + [ + JSON.stringify( + chunk.map((r) => ({ + key: r.key, + name: r.name, + source: r.source, + poster: r.poster, + groups: r.groups, + posted_at: r.postedAt, + file_total: r.fileTotal, + files: r.files, + segments: r.segments, + bytes: r.bytes, + first_article: r.firstArticle, + last_article: r.lastArticle, + readable: r.readable, + has_par2: r.hasPar2, + })) + ), + ] + ) + for (const row of rows) { + ids.set(row.release_key, Number(row.id)) + if (row.inserted) out.newReleases++ + } + } + + // One write-once chunk per release. + for (let i = 0; i < sorted.length; i += 500) { + const rows = sorted.slice(i, i + 500).map((r) => ({ + release_id: ids.get(r.key)!, + range_id: rangeId, + batch_from: batchFrom, + files: r.files, + segments: r.segments, + payload: r.payload, + })) + if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows) + } + + const kept = candidates.filter((c) => ids.has(c.releaseKey)) + for (let i = 0; i < kept.length; i += 2000) { + const rows = kept + .slice(i, i + 2000) + .map((c) => ({ + release_id: ids.get(c.releaseKey)!, + kind: c.kind, + message_id: c.messageId, + file: c.file, + })) + const added = await trx + .knexQuery() + .table('probe_candidates') + .insert(rows) + .onConflict(['release_id', 'kind', 'message_id']) + .ignore() + .returning('id') + out.candidates += added.length + } + + return out + } } export default new BatchStore() diff --git a/app/services/fragment_merger.ts b/app/services/fragment_merger.ts new file mode 100644 index 0000000..3aef3bb --- /dev/null +++ b/app/services/fragment_merger.ts @@ -0,0 +1,117 @@ +import db from '@adonisjs/lucid/services/db' +import BatchStore from '#services/batch_store' +import ChunkPartitions from '#services/chunk_partitions' +import RuleSets from '#services/rule_sets' +import { RuleSet } from '#lib/rules/index' +import { candidatesOf, chunkOf, mergeFragments, releaseKeyOf, type Fragment, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' +import type { AggregatedFile } from '#lib/files/index' +import type { EdgeSegment } from '#lib/edge/index' + +export interface MergeResult { + files: number + completed: number + incomplete: number + waiting: number +} + +/** File keys per round. */ +const BATCH = 2000 + +/** + * Files cut by a range's edge (REBUILD.md, fragments), put together once + * the ranges around them are done: a complete file is judged by the rules + * like any other and written as a release chunk (with probe candidates); a + * file still missing parts while every neighbouring range is done is + * incomplete, and dropped (counted on its ranges). Others wait. + */ +class FragmentMerger { + async run(): Promise { + const result: MergeResult = { files: 0, completed: 0, incomplete: 0, waiting: 0 } + const { rows: keys } = await db.rawQuery( + `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id WHERE r.status = 'done' LIMIT ?`, + [BATCH] + ) + if (keys.length === 0) return result + const rules = new RuleSet(await RuleSets.current()) + const rows = await db + .from('fragments as f') + .join('ranges as r', 'r.id', 'f.range_id') + .whereIn('f.file_key', keys.map((k: { file_key: string }) => k.file_key)) + .select('f.id', 'f.file_key', 'f.range_id', 'f.fragment', 'r.group_id', 'r.start_id', 'r.end_id', 'r.status') + const byKey = new Map() + for (const row of rows) { + const list = byKey.get(row.file_key) + if (list) list.push(row) + else byKey.set(row.file_key, [row]) + } + + const completed: { file: AggregatedFile; rangeId: number; ids: number[] }[] = [] + const incomplete: { rangeId: number; ids: number[]; segments: number }[] = [] + for (const list of byKey.values()) { + result.files++ + const file = mergeFragments(list.map((r) => r.fragment as Fragment)) + const ids = list.map((r) => Number(r.id)) + const rangeId = Math.max(...list.map((r) => Number(r.range_id))) + if (file) { + completed.push({ file, rangeId, ids }) + continue + } + if (list.every((r) => r.status === 'done') && (await this.neighboursDone(list))) { + incomplete.push({ rangeId, ids, segments: list.reduce((n, r) => n + (r.fragment as Fragment).segments.length, 0) }) + } else { + result.waiting++ + } + } + + // Completed files, grouped into their releases, judged and written like a batch's. + const byRelease = new Map[]; rangeId: number }>() + for (const { file, rangeId } of completed) { + const key = releaseKeyOf(file.segments[0]) + const entry = byRelease.get(key) + if (entry) entry.files.push(file) + else byRelease.set(key, { files: [file], rangeId }) + } + const releases: { rangeId: number; chunk: ReleaseChunk; candidates: ProbeCandidate[] }[] = [] + for (const [key, { files, rangeId }] of byRelease) { + const verdict = rules.judge(key, files.map((f) => ({ name: f.segments[0].file.name, totalParts: f.segments[0].totalParts, groups: f.segments[0].file.groups, firstMessageId: f.segments[0].messageId }))) + if (verdict.rule !== null) continue + releases.push({ rangeId, chunk: chunkOf(key, files), candidates: candidatesOf(key, files) }) + } + await ChunkPartitions.ensure() + await db.transaction(async (trx) => { + for (const { rangeId, chunk, candidates } of releases) await BatchStore.writeReleases(trx, rangeId, 0, [chunk], candidates) + const done = [...completed.flatMap((c) => c.ids), ...incomplete.flatMap((c) => c.ids)] + for (let i = 0; i < done.length; i += 5000) await trx.from('fragments').whereIn('id', done.slice(i, i + 5000)).delete() + for (const { rangeId, segments } of incomplete) { + await trx.rawQuery( + `INSERT INTO rule_counts (range_id, rule, key, articles, files) VALUES (?, 'incomplete', 'edge', ?, 1) + ON CONFLICT (range_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + 1`, + [rangeId, segments] + ) + } + }) + result.completed = completed.length + result.incomplete = incomplete.length + return result + } + + /** + * Whether the ranges right before and after these fragments' ranges (where + * a missing part could be) are done. One that isn't planned yet counts as + * not done (newer articles come later), except before the group's oldest article. + */ + private async neighboursDone(list: { group_id: number; start_id: number | string; end_id: number | string }[]) { + const groupId = list[0].group_id + const first = Math.min(...list.map((r) => Number(r.start_id))) + const last = Math.max(...list.map((r) => Number(r.end_id))) + const [before, after, group] = await Promise.all([ + db.from('ranges').where('group_id', groupId).where('end_id', first - 1).select('status').first(), + db.from('ranges').where('group_id', groupId).where('start_id', last + 1).select('status').first(), + db.from('groups').where('id', groupId).select('low').first(), + ]) + const beforeDone = before ? before.status === 'done' : group?.low === null || first <= Number(group?.low ?? 0) + return beforeDone && after?.status === 'done' + } +} + +export default new FragmentMerger() diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index a5a9b5e..a05c6c6 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -3,14 +3,17 @@ import type { CommandOptions } from '@adonisjs/core/types/ace' import { setTimeout as sleep } from 'node:timers/promises' import { Prefetcher } from '#services/prefetcher' import Planner from '#services/planner' +import FragmentMerger from '#services/fragment_merger' import NntpPool from '#services/nntp_pool' const TICK_MS = 2000 const PLAN_EVERY_MS = 60 * 60 * 1000 +const MERGE_EVERY_MS = 60 * 1000 /** * The coordinator's background work (REBUILD.md, part 1): the prefetcher, - * and planning ranges for new articles every hour. The API (leases, stream, + * merging fragments of finished ranges every minute, and planning ranges + * for new articles every hour. The API (leases, stream, * batches) runs in the web server. */ export default class CoordinatorRun extends BaseCommand { @@ -30,6 +33,7 @@ export default class CoordinatorRun extends BaseCommand { await done }) let plannedAt = 0 + let mergedAt = 0 this.logger.info('Coordinator started.') while (!stopping) { try { @@ -38,6 +42,11 @@ export default class CoordinatorRun extends BaseCommand { const { groups, ranges } = await Planner.planAll() if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`) } + if (Date.now() - mergedAt >= MERGE_EVERY_MS) { + mergedAt = Date.now() + const merged = await FragmentMerger.run() + if (merged.completed || merged.incomplete) this.logger.info(`Fragments: ${merged.completed} files completed, ${merged.incomplete} incomplete, ${merged.waiting} waiting.`) + } await prefetcher.tick() } catch (error: any) { this.logger.error(`Coordinator round failed: ${error.message}`) diff --git a/lib/edge/index.ts b/lib/edge/index.ts index 6d1c751..8663d2a 100644 --- a/lib/edge/index.ts +++ b/lib/edge/index.ts @@ -1,4 +1,6 @@ -export { RangeIndexer, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' +export { RangeIndexer, chunkOf, candidatesOf, releaseKeyOf, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js' export { toWire, fromWire, type WireBatch } from './wire.js' export type { Lease } from './lease.js' +export { mergeFragments } from './merge.js' +export type { EdgeSegment, FileMeta } from './scan.js' diff --git a/lib/edge/merge.ts b/lib/edge/merge.ts new file mode 100644 index 0000000..87c64bf --- /dev/null +++ b/lib/edge/merge.ts @@ -0,0 +1,35 @@ +import type { AggregatedFile } from '../files/index.js' +import type { Fragment } from './range_indexer.js' +import type { EdgeSegment, FileMeta } from './scan.js' + +/** + * The fragments of one file from neighbouring ranges, put together + * (REBUILD.md, fragments): the complete file when every part from 1 to its + * total is there (the first article seen of a part wins), else null. + */ +export function mergeFragments(fragments: Fragment[]): AggregatedFile | null { + if (fragments.length === 0) return null + const first = fragments[0] + const meta: FileMeta = { + fileKey: first.fileKey, + name: first.name, + poster: first.poster, + subject: first.subject, + totalParts: first.totalParts, + groups: first.groups, + release: first.release ? { ...first.release, source: first.release.source as any } : null, + } + const parts = new Map() + for (const fragment of fragments) { + const postedAt = fragment.postedAt ?? Number.NaN + for (const [number, messageId, bytes, articleNumber] of fragment.segments) { + const known = parts.get(number) + if (!known || articleNumber < known.articleNumber) { + parts.set(number, { fileKey: meta.fileKey, number, totalParts: meta.totalParts, messageId, bytes, articleNumber, postedAt, file: meta }) + } + } + } + if (parts.size < meta.totalParts) return null + for (let n = 1; n <= meta.totalParts; n++) if (!parts.has(n)) return null + return { fileKey: meta.fileKey, segments: [...parts.values()].sort((a, b) => a.number - b.number) } +} diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index c285aca..e8b2d7c 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -280,7 +280,7 @@ export class RangeIndexer { } /** The release key; a file without a recognisable release becomes a release of its own. */ -function releaseKeyOf(s: EdgeSegment): string { +export function releaseKeyOf(s: EdgeSegment): string { return s.file.release?.key ?? createHash('sha1').update(`file\n${s.fileKey}`).digest('hex') } @@ -293,7 +293,7 @@ function postedOf(segments: EdgeSegment[]): number | null { return min } -function chunkOf(key: string, files: AggregatedFile[]): ReleaseChunk { +export function chunkOf(key: string, files: AggregatedFile[]): ReleaseChunk { const first = files[0].segments[0].file const chunkFiles: ChunkFile[] = files.map((f) => { const m = f.segments[0].file @@ -339,7 +339,7 @@ function chunkOf(key: string, files: AggregatedFile[]): ReleaseChun * volume, of its par2 index, of a lone random-named file, and a 7z set's * last volume (its listing is in the tail). At most a few per release. */ -function candidatesOf(key: string, files: AggregatedFile[]): ProbeCandidate[] { +export function candidatesOf(key: string, files: AggregatedFile[]): ProbeCandidate[] { const out: ProbeCandidate[] = [] let tail: { volume: number; file: AggregatedFile } | null = null for (const f of files) { diff --git a/tests/functional/fragment_merger.spec.ts b/tests/functional/fragment_merger.spec.ts new file mode 100644 index 0000000..9bc2e20 --- /dev/null +++ b/tests/functional/fragment_merger.spec.ts @@ -0,0 +1,65 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import BatchStore from '#services/batch_store' +import FragmentMerger from '#services/fragment_merger' +import { RangeIndexer } from '#lib/edge/index' +import { decodeChunk } from '#lib/chunks/index' +import { DEFAULT_RULES } from '#lib/rules/index' +import type { Overview } from '#lib/nntp/index' + +const GROUP = 'alt.binaries.test' +const rules = { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } } + +function parts(name: string, total: number, numbers: number[], at: number): Overview[] { + return numbers.map((n, i) => [at + i, { + 'subject': `Big.Upload.2020 [1/1] - "${name}" yEnc (${n}/${total})`, 'from': 'p ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', + 'message-id': `<${name}-${n}@test>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}`, + }] as Overview) +} + +/** Indexes one range's overviews and stores them as the range's only batch; the range ends done. */ +async function indexRange(groupId: number, nodeId: number, start: number, end: number, overviews: Overview[]) { + const [r] = await db.table('ranges').insert({ group_id: groupId, start_id: start, end_id: end, next_id: start, status: 'leased', node_id: nodeId }).returning('id') + const indexer = new RangeIndexer({ group: GROUP, start, end, rules, edgeArticles: 100 }) + indexer.add(overviews, end + 1) + indexer.finish() + await BatchStore.store(Number(r.id), nodeId, indexer.take()) + await db.from('ranges').where('id', r.id).update({ status: 'done' }) + return Number(r.id) +} + +test.group('FragmentMerger', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('a file cut by a range edge is completed from both sides once both ranges are done', async ({ assert }) => { + const [g] = await db.table('groups').insert({ name: GROUP, priority: 1, low: 1 }).returning('id') + const [n] = await db.table('nodes').insert({ name: 'merge-node' }).returning('id') + // Parts 1–2 at the end of the first range (still open there), 3–4 at the start of the next. + await indexRange(g.id, n.id, 1, 1000, parts('big.part1.rar', 4, [1, 2], 990)) + assert.equal((await db.from('fragments').count('* as c'))[0].c, '1') + assert.include(await FragmentMerger.run(), { completed: 0, incomplete: 0, waiting: 1 }) + + await indexRange(g.id, n.id, 1001, 2000, parts('big.part1.rar', 4, [3, 4], 1001)) + assert.include(await FragmentMerger.run(), { completed: 1, incomplete: 0 }) + assert.equal((await db.from('fragments').count('* as c'))[0].c, '0') + const release = await db.from('releases').first() + assert.include(release, { files: 1, chunks: 1 }) + const chunk = await db.from('release_chunks').where('release_id', release.id).first() + assert.deepEqual(decodeChunk(chunk.payload)[0].segments.map((s) => s[0]), [1, 2, 3, 4]) + assert.deepEqual((await db.from('probe_candidates')).map((c) => c.kind), ['archive']) + }) + + test('a file still missing parts when all its neighbours are done is dropped', async ({ assert }) => { + const [g] = await db.table('groups').insert({ name: GROUP, priority: 1, low: 1 }).returning('id') + const [n] = await db.table('nodes').insert({ name: 'merge-node' }).returning('id') + await indexRange(g.id, n.id, 1, 1000, parts('gap.part1.rar', 5, [1, 2], 990)) + await indexRange(g.id, n.id, 1001, 2000, parts('gap.part1.rar', 5, [4, 5], 1001)) + // Its later neighbour isn't there yet: the missing part could still come. + assert.include(await FragmentMerger.run(), { completed: 0, incomplete: 0, waiting: 1 }) + await indexRange(g.id, n.id, 2001, 3000, []) + assert.include(await FragmentMerger.run(), { completed: 0, incomplete: 1 }) + assert.equal((await db.from('fragments').count('* as c'))[0].c, '0') + assert.lengthOf(await db.from('releases'), 0) + }) +})