| 1 | import { |
| 2 | createAbortError, |
| 3 | ResourceLock, |
| 4 | resourceClaimsConflict, |
| 5 | type ReleaseFunc |
| 6 | } from '../lock/resource-lock' |
| 7 | import type { |
| 8 | ActiveJob, |
| 9 | JobLease, |
| 10 | JobOwner, |
| 11 | JobReservationArgs, |
| 12 | JobReservationResult |
| 13 | } from './types' |
| 14 | |
| 15 | type ManagedJob = ActiveJob & { |
| 16 | ownerToken: symbol |
| 17 | controller: AbortController |
| 18 | releaseLock?: ReleaseFunc |
| 19 | removeExternalAbortListener: () => void |
| 20 | released: boolean |
| 21 | } |
| 22 | |
| 23 | const ownerKey = (owner: JobOwner): string => `${owner.kind}:${owner.id}` |
| 24 | |
| 25 | const relayAbort = (source: AbortSignal | undefined, target: AbortController): (() => void) => { |
| 26 | if (!source) return () => undefined |
| 27 | const abort = (): void => target.abort(source.reason) |
| 28 | if (source.aborted) { |
| 29 | abort() |
| 30 | return () => undefined |
| 31 | } |
| 32 | source.addEventListener('abort', abort, { once: true }) |
| 33 | return () => source.removeEventListener('abort', abort) |
| 34 | } |
| 35 | |
| 36 | /** The sole owner of resource claims and run-level cancellation for Runtime jobs. */ |
| 37 | export class JobCoordinator { |
| 38 | private readonly lock: ResourceLock |
| 39 | private readonly jobsById = new Map<string, ManagedJob>() |
| 40 | private readonly jobIdByOwner = new Map<string, string>() |
| 41 | |
| 42 | constructor(lock = new ResourceLock()) { |
| 43 | this.lock = lock |
| 44 | } |
| 45 | |
| 46 | async reserve(args: JobReservationArgs): Promise<JobReservationResult> { |
| 47 | if (args.wait === 'fail') return this.tryReserve(args) |
| 48 | if (this.jobsById.has(args.jobId)) { |
| 49 | return { status: 'busy', conflictingJobId: args.jobId } |
| 50 | } |
| 51 | const existingJobId = this.jobIdByOwner.get(ownerKey(args.owner)) |
| 52 | if (existingJobId) return { status: 'busy', conflictingJobId: existingJobId } |
| 53 | if (args.signal?.aborted) throw createAbortError() |
| 54 | |
| 55 | const controller = new AbortController() |
| 56 | const job: ManagedJob = { |
| 57 | jobId: args.jobId, |
| 58 | domain: args.domain, |
| 59 | owner: args.owner, |
| 60 | state: 'waiting', |
| 61 | claims: args.claims, |
| 62 | ownerToken: Symbol(args.jobId), |
| 63 | controller, |
| 64 | removeExternalAbortListener: relayAbort(args.signal, controller), |
| 65 | released: false |
| 66 | } |
| 67 | this.jobsById.set(job.jobId, job) |
| 68 | this.jobIdByOwner.set(ownerKey(job.owner), job.jobId) |
| 69 | |
| 70 | try { |
| 71 | const releaseLock = await this.lock.acquire(job.claims, { |
| 72 | ownerToken: job.ownerToken, |
| 73 | signal: controller.signal, |
| 74 | wait: args.wait |
| 75 | }) |
| 76 | if (!releaseLock) { |
| 77 | const conflictingJobId = this.findConflictingJobId(job) |
| 78 | this.removeJob(job) |
| 79 | if (!conflictingJobId) { |
| 80 | throw new Error('ResourceLock reported a conflict without a registered Runtime job') |
| 81 | } |
| 82 | return { status: 'busy', conflictingJobId } |
| 83 | } |
| 84 | |
| 85 | job.releaseLock = releaseLock |
| 86 | job.state = 'active' |
| 87 | return { |
| 88 | status: 'acquired', |
| 89 | lease: this.createLease(job) |
| 90 | } |
| 91 | } catch (error) { |
| 92 | this.removeJob(job) |
| 93 | throw error |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | /** |
| 98 | * Non-waiting variant for old synchronous IPC handlers. It keeps the same |
| 99 | * owner map, claims, and cancellation controller as async reserve(). |
| 100 | */ |
| 101 | tryReserve(args: JobReservationArgs): JobReservationResult { |
| 102 | if (args.wait !== 'fail') { |
| 103 | throw new Error('JobCoordinator.tryReserve only supports wait=fail') |
| 104 | } |
| 105 | if (this.jobsById.has(args.jobId)) { |
| 106 | return { status: 'busy', conflictingJobId: args.jobId } |
| 107 | } |
| 108 | const existingJobId = this.jobIdByOwner.get(ownerKey(args.owner)) |
| 109 | if (existingJobId) return { status: 'busy', conflictingJobId: existingJobId } |
| 110 | if (args.signal?.aborted) throw createAbortError() |
| 111 | |
| 112 | const controller = new AbortController() |
| 113 | const job: ManagedJob = { |
| 114 | jobId: args.jobId, |
| 115 | domain: args.domain, |
| 116 | owner: args.owner, |
| 117 | state: 'waiting', |
| 118 | claims: args.claims, |
| 119 | ownerToken: Symbol(args.jobId), |
| 120 | controller, |
| 121 | removeExternalAbortListener: relayAbort(args.signal, controller), |
| 122 | released: false |
| 123 | } |
| 124 | this.jobsById.set(job.jobId, job) |
| 125 | this.jobIdByOwner.set(ownerKey(job.owner), job.jobId) |
| 126 | |
| 127 | try { |
| 128 | const releaseLock = this.lock.tryAcquire(job.claims, { |
| 129 | ownerToken: job.ownerToken, |
| 130 | signal: controller.signal |
| 131 | }) |
| 132 | if (!releaseLock) { |
| 133 | const conflictingJobId = this.findConflictingJobId(job) |
| 134 | this.removeJob(job) |
| 135 | if (!conflictingJobId) { |
| 136 | throw new Error('ResourceLock reported a conflict without a registered Runtime job') |
| 137 | } |
| 138 | return { status: 'busy', conflictingJobId } |
| 139 | } |
| 140 | |
| 141 | job.releaseLock = releaseLock |
| 142 | job.state = 'active' |
| 143 | return { status: 'acquired', lease: this.createLease(job) } |
| 144 | } catch (error) { |
| 145 | this.removeJob(job) |
| 146 | throw error |
| 147 | } |
| 148 | } |
| 149 | |
| 150 | cancel(jobId: string): boolean { |
| 151 | const job = this.jobsById.get(jobId) |
| 152 | if (!job || job.controller.signal.aborted) return false |
| 153 | job.controller.abort() |
| 154 | return true |
| 155 | } |
| 156 | |
| 157 | cancelOwner(owner: JobOwner): number { |
| 158 | let cancelled = 0 |
| 159 | for (const job of this.jobsById.values()) { |
| 160 | if (job.owner.kind === owner.kind && job.owner.id === owner.id && this.cancel(job.jobId)) { |
| 161 | cancelled += 1 |
| 162 | } |
| 163 | } |
| 164 | return cancelled |
| 165 | } |
| 166 | |
| 167 | getByOwner(owner: JobOwner): ActiveJob | null { |
| 168 | const jobId = this.jobIdByOwner.get(ownerKey(owner)) |
| 169 | const job = jobId ? this.jobsById.get(jobId) : undefined |
| 170 | return job ? this.toActiveJob(job) : null |
| 171 | } |
| 172 | |
| 173 | private createLease(job: ManagedJob): JobLease { |
| 174 | return { |
| 175 | jobId: job.jobId, |
| 176 | signal: job.controller.signal, |
| 177 | release: () => this.release(job) |
| 178 | } |
| 179 | } |
| 180 | |
| 181 | private release(job: ManagedJob): void { |
| 182 | if (job.released) return |
| 183 | job.released = true |
| 184 | job.releaseLock?.() |
| 185 | this.removeJob(job) |
| 186 | } |
| 187 | |
| 188 | private removeJob(job: ManagedJob): void { |
| 189 | job.removeExternalAbortListener() |
| 190 | if (this.jobsById.get(job.jobId) === job) this.jobsById.delete(job.jobId) |
| 191 | if (this.jobIdByOwner.get(ownerKey(job.owner)) === job.jobId) { |
| 192 | this.jobIdByOwner.delete(ownerKey(job.owner)) |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | private findConflictingJobId(job: ManagedJob): string | undefined { |
| 197 | for (const candidate of this.jobsById.values()) { |
| 198 | if (candidate === job) continue |
| 199 | if (resourceClaimsConflict(job.claims, candidate.claims)) return candidate.jobId |
| 200 | } |
| 201 | return undefined |
| 202 | } |
| 203 | |
| 204 | private toActiveJob(job: ManagedJob): ActiveJob { |
| 205 | return { |
| 206 | jobId: job.jobId, |
| 207 | domain: job.domain, |
| 208 | owner: job.owner, |
| 209 | state: job.state, |
| 210 | claims: job.claims |
| 211 | } |
| 212 | } |
| 213 | } |
| 214 |