SolidScribe/server/tests/helpers/HostQueue.test.js

require('module-alias/register')

const HostQueue = require('@helpers/HostQueue')

//
// A controllable clock that also drives the queue's own timers, so advancing
// time is instant and spacing assertions are exact rather than flaky.
//
const makeClock = () => {
	let t = 1000
	let nextId = 1
	const timers = new Map()

	return {
		now: () => t,
		setTimeoutFn: (fn, ms) => {
			const id = nextId++
			timers.set(id, { fn, dueAt: t + ms })
			return id
		},
		clearTimeoutFn: (id) => { timers.delete(id) },
		advance: (ms) => {
			t += ms
			for (const [id, timer] of [...timers.entries()]) {
				if (timer.dueAt <= t) {
					timers.delete(id)
					timer.fn()
				}
			}
		},
	}
}

const wireClock = (clock) => ({
	now: clock.now,
	setTimeoutFn: clock.setTimeoutFn,
	clearTimeoutFn: clock.clearTimeoutFn,
})

const deferred = () => {
	let resolve = null
	let reject = null
	const promise = new Promise((res, rej) => { resolve = res; reject = rej })
	return { promise, resolve, reject }
}

describe('global concurrency', () => {

	test('never exceeds the cap', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 2, minSpacingMs: 0, maxSpacingMs: 0 })

		let active = 0
		let peak = 0
		const gates = []

		const jobs = Array.from({ length: 6 }, (unused, i) => {
			const gate = deferred()
			gates.push(gate)
			return queue.schedule(`host${i}.com`, async () => {
				active++
				peak = Math.max(peak, active)
				await gate.promise
				active--
				return i
			})
		})

		// Let the first batch start.
		await new Promise(resolve => setImmediate(resolve))
		expect(peak).toBe(2)

		gates.forEach(gate => gate.resolve())
		await Promise.all(jobs)

		expect(peak).toBe(2)
	})

	test('runs different hosts in parallel up to the cap', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 4, minSpacingMs: 0, maxSpacingMs: 0 })

		let active = 0
		let peak = 0
		const gates = []

		const jobs = ['a.com', 'b.com', 'c.com'].map(host => {
			const gate = deferred()
			gates.push(gate)
			return queue.schedule(host, async () => {
				active++
				peak = Math.max(peak, active)
				await gate.promise
				active--
			})
		})

		await new Promise(resolve => setImmediate(resolve))
		expect(peak).toBe(3)

		gates.forEach(gate => gate.resolve())
		await Promise.all(jobs)
	})
})

describe('per-host serialisation', () => {

	test('runs only one request per host at a time', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 4, minSpacingMs: 0, maxSpacingMs: 0 })

		let active = 0
		let peak = 0

		const jobs = [1, 2, 3].map(() => queue.schedule('same.com', async () => {
			active++
			peak = Math.max(peak, active)
			await new Promise(resolve => setImmediate(resolve))
			active--
		}))

		await Promise.all(jobs)

		expect(peak).toBe(1)
	})

	test('preserves FIFO order within a host', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 1, minSpacingMs: 0, maxSpacingMs: 0 })
		const order = []

		await Promise.all([1, 2, 3, 4].map(n =>
			queue.schedule('same.com', async () => { order.push(n) })
		))

		expect(order).toEqual([1, 2, 3, 4])
	})
})

describe('politeness spacing', () => {

	test('leaves a gap between two requests to the same host', async () => {
		const clock = makeClock()
		const queue = new HostQueue.Queue({
			globalConcurrency: 4,
			minSpacingMs: 1000,
			maxSpacingMs: 1000,
			...wireClock(clock),
		})

		const startedAt = []

		const first = queue.schedule('same.com', async () => { startedAt.push(clock.now()) })
		await first

		const second = queue.schedule('same.com', async () => { startedAt.push(clock.now()) })

		// The second job must not have run yet - the host is spaced out.
		await new Promise(resolve => setImmediate(resolve))
		expect(startedAt).toHaveLength(1)

		clock.advance(1000)
		await second

		expect(startedAt).toHaveLength(2)
		expect(startedAt[1] - startedAt[0]).toBeGreaterThanOrEqual(1000)
	})

	test('spacing on one host does not delay a different host', async () => {
		const clock = makeClock()
		const queue = new HostQueue.Queue({
			globalConcurrency: 4,
			minSpacingMs: 5000,
			maxSpacingMs: 5000,
			...wireClock(clock),
		})

		await queue.schedule('slow.com', async () => {})

		// slow.com is now parked for 5s. other.com must be unaffected.
		const ran = []
		await queue.schedule('other.com', async () => { ran.push('other') })

		expect(ran).toEqual(['other'])
	})

	test('honours an explicit spacing override (robots Crawl-delay)', async () => {
		const clock = makeClock()
		const queue = new HostQueue.Queue({ minSpacingMs: 0, maxSpacingMs: 0, ...wireClock(clock) })

		await queue.schedule('slow.com', async () => {}, { spacingMs: 3000 })

		const ran = []
		const second = queue.schedule('slow.com', async () => { ran.push(1) })

		await new Promise(resolve => setImmediate(resolve))
		expect(ran).toEqual([])

		clock.advance(3000)
		await second
		expect(ran).toEqual([1])
	})
})

describe('error handling', () => {

	test('a rejecting task does not wedge the queue', async () => {
		const queue = new HostQueue.Queue({ minSpacingMs: 0, maxSpacingMs: 0 })

		await expect(queue.schedule('a.com', async () => { throw new Error('boom') }))
			.rejects.toThrow('boom')

		await expect(queue.schedule('a.com', async () => 'fine')).resolves.toBe('fine')
	})

	test('a task that throws synchronously is caught', async () => {
		const queue = new HostQueue.Queue({ minSpacingMs: 0, maxSpacingMs: 0 })

		await expect(queue.schedule('a.com', () => { throw new Error('sync boom') }))
			.rejects.toThrow('sync boom')

		expect(queue.inflight()).toBe(0)
	})

	test('one failing host does not block others', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 2, minSpacingMs: 0, maxSpacingMs: 0 })

		const results = await Promise.allSettled([
			queue.schedule('bad.com', async () => { throw new Error('nope') }),
			queue.schedule('good.com', async () => 'ok'),
		])

		expect(results[0].status).toBe('rejected')
		expect(results[1].value).toBe('ok')
	})
})

describe('drain and stop', () => {

	test('drain resolves when everything has finished', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 2, minSpacingMs: 0, maxSpacingMs: 0 })
		const done = []

		;[1, 2, 3].forEach(n => queue.schedule(`h${n}.com`, async () => { done.push(n) }).catch(() => {}))

		await queue.drain()

		expect(done).toHaveLength(3)
		expect(queue.size()).toBe(0)
		expect(queue.inflight()).toBe(0)
	})

	test('drain resolves immediately on an idle queue', async () => {
		const queue = new HostQueue.Queue()
		await expect(queue.drain()).resolves.toBeUndefined()
	})

	test('stop rejects queued work and refuses new work', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 1, minSpacingMs: 0, maxSpacingMs: 0 })

		const gate = deferred()
		const first = queue.schedule('a.com', () => gate.promise)
		const queued = queue.schedule('b.com', async () => 'never')

		await new Promise(resolve => setImmediate(resolve))
		queue.stop()

		await expect(queued).rejects.toThrow('stopped')
		await expect(queue.schedule('c.com', async () => 'x')).rejects.toThrow('stopped')

		gate.resolve('done')
		await expect(first).resolves.toBe('done')
	})
})

describe('event loop', () => {

	//
	// Regression: the politeness timer used to be unref'd, so if nothing else
	// held the event loop open Node exited and queued jobs were abandoned with
	// their promises never settling.
	//
	test('a spaced job still resolves with nothing else holding the loop open', async () => {
		const queue = new HostQueue.Queue({ minSpacingMs: 25, maxSpacingMs: 25 })

		await queue.schedule('same.com', async () => 'first')

		// Only the internal timer can resume this one.
		await expect(queue.schedule('same.com', async () => 'second')).resolves.toBe('second')
	})

	test('stop() clears the pending timer so shutdown is not blocked', async () => {
		const queue = new HostQueue.Queue({ minSpacingMs: 10000, maxSpacingMs: 10000 })

		await queue.schedule('same.com', async () => 'first')
		const queued = queue.schedule('same.com', async () => 'never')

		queue.stop()

		await expect(queued).rejects.toThrow('stopped')
	})
})

describe('stats', () => {

	test('reports pending, running and host counts', async () => {
		const queue = new HostQueue.Queue({ globalConcurrency: 1, minSpacingMs: 0, maxSpacingMs: 0 })

		const gate = deferred()
		const first = queue.schedule('a.com', () => gate.promise)
		const second = queue.schedule('b.com', async () => {})

		await new Promise(resolve => setImmediate(resolve))

		const stats = queue.stats()
		expect(stats.running).toBe(1)
		expect(stats.pending).toBe(1)

		gate.resolve()
		await Promise.all([first, second])
	})
})