Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ follows this truth table:
| Yes | empty | empty | **Ok** -- worker arriving late. Queue already drained, nothing to do (exit 0). |
| Yes | populated | empty | **Ok. Classic run** -- steal from the shared queue, recording each batch to the rerun key when `--key-rerun` is set. |
| Yes | empty | populated | **Ok. Classic rerun** -- replay exactly the recorded files, ignoring the shared queue. |
| Yes | populated | populated | **Crash (exit 1)** -- weird state, refuse to run. |
| Yes | populated | populated | **Ok. Full rerun** -- the queue was re-pushed while this runner still holds rerun memory from a previous run. The stale rerun key is deleted and the runner steals from the shared queue like a classic run, re-recording as it goes. |

### Complete GitHub Actions example with re-run support

Expand Down Expand Up @@ -250,7 +250,8 @@ jobs:
- **Push** uses `RPUSH` to append all file paths to a Redis list in a single command, then sets `EXPIRE` on the key (default: 1 week). It also sets a `<key>:published` marker with the same TTL so workers can tell a drained queue from one that was never published.
- **Steal** uses `LPOP key count` (Redis 6.2+), which atomically pops up to N elements. No Lua scripts, no locks, no race conditions.
- **Record** (when `--key-rerun` is set): after each steal, the batch is also `RPUSH`ed to the per-runner rerun key with the same TTL.
- **Replay** (when `--key-rerun` has data): reads all files from the rerun key via `LRANGE` (non-destructive), splits into batches, and runs them locally.
- **Replay** (when `--key-rerun` has data and the shared queue is drained): reads all files from the rerun key via `LRANGE` (non-destructive), splits into batches, and runs them locally.
- **Full rerun** (when both the shared queue and the rerun key have data): the stale rerun key is removed with `DEL`, then the runner steals from the shared queue and re-records as in a classic run.
- **Run** spawns the configured command via `child_process.spawnSync()` with file paths as arguments. No shell expansion overhead.
- **Exit code** is 0 if every batch passed (or the queue was already empty), 1 if any batch had failures.

Expand Down
10 changes: 10 additions & 0 deletions src/redisQueue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,16 @@ export class RedisQueue {
})
}

/**
* Remove a key entirely. Used to discard stale rerun memory when a
* full rerun starts over from the shared queue.
*/
async delete(key: string): Promise<number> {
return this.withRetries('delete', async () => {
return this.redis.del(key)
})
}

async close(): Promise<void> {
await this.redis.quit()
}
Expand Down
14 changes: 10 additions & 4 deletions src/worker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ export class Worker {
* Yes | empty | empty | Ok — worker arriving late
* Yes | populated | empty | Ok — classic run (steal + record)
* Yes | empty | populated | Ok — classic rerun (replay)
* Yes | populated | populated | Crashweird state, refuse
* Yes | populated | populated | Okfull rerun (reset rerun key, steal)
*
* Returns 0 if all batches passed (or nothing to do), 1 on failure or crash.
*/
Expand All @@ -107,10 +107,16 @@ export class Worker {
if (rerunFiles.length > 0) {
const queueLen = await this.queue.length(this.key)
if (queueLen > 0) {
this.log(`[specbandit] ERROR: weird state — shared queue '${this.key}' (${queueLen} files) and rerun key '${this.keyRerun}' (${rerunFiles.length} files) both have data. Refusing to run.`)
return 1
// The queue was re-pushed while this runner still carries rerun
// memory from a previous run (a full rerun). The stored memory is
// stale — discard it and steal from the shared queue like a classic
// run, re-recording each stolen batch as we go.
this.log(`[specbandit] Shared queue '${this.key}' and rerun key '${this.keyRerun}' both have files. Full rerun: resetting '${this.keyRerun}' and working from '${this.key}'.`)
await this.queue.delete(this.keyRerun!)
exitCode = await this.runSteal(true)
} else {
exitCode = await this.runReplay(rerunFiles)
}
exitCode = await this.runReplay(rerunFiles)
} else {
// No rerun data: steal from the shared queue, recording each batch to
// the rerun key when one is configured. Handles both the classic run
Expand Down
13 changes: 13 additions & 0 deletions test/redisQueue.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ describe('RedisQueue', () => {
lrange: ReturnType<typeof vi.fn>
set: ReturnType<typeof vi.fn>
exists: ReturnType<typeof vi.fn>
del: ReturnType<typeof vi.fn>
quit: ReturnType<typeof vi.fn>
}

Expand All @@ -26,6 +27,7 @@ describe('RedisQueue', () => {
lrange: vi.fn(),
set: vi.fn(),
exists: vi.fn(),
del: vi.fn(),
quit: vi.fn(),
}

Expand Down Expand Up @@ -133,6 +135,17 @@ describe('RedisQueue', () => {
})
})

describe('#delete', () => {
it('removes the key via DEL', async () => {
mockRedis.del.mockResolvedValue(1)

const result = await queue.delete('my-key')

expect(mockRedis.del).toHaveBeenCalledWith('my-key')
expect(result).toBe(1)
})
})

describe('#markPublished', () => {
it('sets the published marker key with a TTL via SET EX', async () => {
mockRedis.set.mockResolvedValue('OK')
Expand Down
68 changes: 50 additions & 18 deletions test/worker.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ function createMockQueue() {
steal: vi.fn().mockResolvedValue([]),
length: vi.fn().mockResolvedValue(0),
readAll: vi.fn().mockResolvedValue([]),
delete: vi.fn().mockResolvedValue(1),
isPublished: vi.fn().mockResolvedValue(true),
markPublished: vi.fn().mockResolvedValue(undefined),
close: vi.fn().mockResolvedValue(undefined),
Expand Down Expand Up @@ -555,6 +556,7 @@ describe('Worker adapter lifecycle', () => {
steal: ReturnType<typeof vi.fn>
readAll: ReturnType<typeof vi.fn>
length: ReturnType<typeof vi.fn>
delete: ReturnType<typeof vi.fn>
isPublished: ReturnType<typeof vi.fn>
}
let capture: ReturnType<typeof createOutputCapture>
Expand Down Expand Up @@ -671,28 +673,58 @@ describe('Worker adapter lifecycle', () => {
expect(adapter.runBatch).not.toHaveBeenCalled()
})

it('crashes (exit 1) on the weird case: both shared queue and rerun key have data', async () => {
queue.isPublished.mockResolvedValue(true)
queue.readAll.mockResolvedValue(['test/x.test.ts'])
queue.length.mockResolvedValue(3) // shared queue still populated
const adapter = createMockAdapter()
describe('full rerun: both shared queue and rerun key have data', () => {
beforeEach(() => {
queue.isPublished.mockResolvedValue(true)
queue.readAll.mockResolvedValue(['test/x.test.ts']) // stale rerun memory
queue.length.mockResolvedValue(3) // shared queue re-pushed
queue.steal
.mockResolvedValueOnce(['test/a.test.ts', 'test/b.test.ts'])
.mockResolvedValueOnce([])
})

const worker = new Worker({
key,
adapter,
batchSize: 2,
keyRerun,
keyTtl: 604_800,
queue: queue as unknown as RedisQueue,
output: capture.stream,
function createWorker(adapter: Adapter) {
return new Worker({
key,
adapter,
batchSize: 2,
keyRerun,
keyTtl: 604_800,
queue: queue as unknown as RedisQueue,
output: capture.stream,
})
}

it('resets the rerun key before stealing from the shared queue', async () => {
const adapter = createMockAdapter()

const exitCode = await createWorker(adapter).run()

expect(exitCode).toBe(0)
expect(queue.delete).toHaveBeenCalledWith(keyRerun)
expect(queue.delete.mock.invocationCallOrder[0]).toBeLessThan(
queue.steal.mock.invocationCallOrder[0],
)
expect(adapter.runBatch).toHaveBeenCalledWith(['test/a.test.ts', 'test/b.test.ts'], 1)
})

const exitCode = await worker.run()
it('re-records the stolen batches to the rerun key', async () => {
await createWorker(createMockAdapter()).run()

expect(exitCode).toBe(1)
expect(capture.getOutput()).toContain('weird state')
expect(queue.steal).not.toHaveBeenCalled()
expect(adapter.runBatch).not.toHaveBeenCalled()
expect(queue.push).toHaveBeenCalledWith(
keyRerun,
['test/a.test.ts', 'test/b.test.ts'],
604_800,
)
})

it('explains the full-rerun reset instead of crashing', async () => {
await createWorker(createMockAdapter()).run()

expect(capture.getOutput()).toContain('Full rerun')
expect(capture.getOutput()).toContain(`resetting '${keyRerun}'`)
expect(capture.getOutput()).not.toContain('ERROR')
})
})

it('exits 0 (nothing to do) when a worker arrives late: published but both keys empty', async () => {
Expand Down
Loading