diff --git a/server/src/__tests__/admission-control-service.test.ts b/server/src/__tests__/admission-control-service.test.ts index c8ec521c..73bb8459 100644 --- a/server/src/__tests__/admission-control-service.test.ts +++ b/server/src/__tests__/admission-control-service.test.ts @@ -14,6 +14,7 @@ import { AdmissionControlService, type AdmissionControlServiceOptions, } from '../services/admission-control-service.js'; +import { acquireLock } from '../services/file-lock.js'; import { FileAdmissionReservationRepository } from '../storage/admission-reservation-repository.js'; import { SqliteDatabase } from '../storage/sqlite/database.js'; import { SqliteAdmissionReservationRepository } from '../storage/sqlite/admission-reservation-repository.js'; @@ -422,6 +423,60 @@ describe('AdmissionControlService', () => { ).rejects.toThrow(/no forward progress/i); }); + it('does not expose an incomplete admission snapshot to concurrent readers', async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'veritas-admission-read-lock-')); + roots.push(root); + const logPath = path.join(root, 'admission.jsonl'); + const repository = new FileAdmissionReservationRepository(logPath); + const service = createService(repository, configuredSettings()); + await service.admit(request('task-read-lock')); + const completeLog = await fs.readFile(logPath); + const splitAt = Math.floor(completeLog.byteLength / 2); + const unlock = await acquireLock(logPath); + const handle = await fs.open(logPath, 'w'); + let handleClosed = false; + let read: + | Promise<{ + records: Awaited> | null; + error: unknown; + }> + | undefined; + + try { + await handle.write(completeLog.subarray(0, splitAt)); + await handle.sync(); + let readSettled = false; + read = repository + .list({ taskId: 'task-read-lock' }) + .then( + (records) => ({ records, error: null }), + (error: unknown) => ({ records: null, error }) + ) + .finally(() => { + readSettled = true; + }); + + await new Promise((resolve) => setImmediate(resolve)); + expect(readSettled).toBe(false); + await handle.write(completeLog.subarray(splitAt)); + await handle.sync(); + await handle.close(); + handleClosed = true; + } finally { + if (!handleClosed) await handle.close().catch(() => {}); + await unlock(); + } + + await expect(read).resolves.toMatchObject({ + records: [ + expect.objectContaining({ + request: expect.objectContaining({ taskId: 'task-read-lock' }), + }), + ], + error: null, + }); + }); + it('renews a leased queue claim until dispatch ownership takes over', async () => { const repository = await repositoryFor('file'); const settings = { diff --git a/server/src/storage/admission-reservation-repository.ts b/server/src/storage/admission-reservation-repository.ts index fbd0256c..0ad31ffe 100644 --- a/server/src/storage/admission-reservation-repository.ts +++ b/server/src/storage/admission-reservation-repository.ts @@ -273,65 +273,77 @@ export class FileAdmissionReservationRepository implements AdmissionReservationR } async get(id: string): Promise { - return this.materialize(await this.readSnapshots()).get(id) ?? null; + await this.prepareParent(); + return withFileLock( + this.filePath, + async () => this.materialize(await this.readSnapshots()).get(id) ?? null + ); } async list(query: AdmissionReservationListQuery): Promise { - const states = query.states ? new Set(query.states) : undefined; - return [...this.materialize(await this.readSnapshots()).values()] - .filter((record) => !query.workspaceId || record.request.workspaceId === query.workspaceId) - .filter((record) => !query.taskId || record.request.taskId === query.taskId) - .filter((record) => !query.rootTaskId || record.request.rootTaskId === query.rootTaskId) - .filter((record) => !query.provider || record.request.provider === query.provider) - .filter((record) => !query.hostId || record.request.hostId === query.hostId) - .filter( - (record) => !query.workflowRunId || record.request.workflowRunId === query.workflowRunId - ) - .filter( - (record) => !query.workflowStepId || record.request.workflowStepId === query.workflowStepId - ) - .filter( - (record) => - !query.rootReservationId || record.request.rootReservationId === query.rootReservationId - ) - .filter( - (record) => - !query.rootObjectiveId || - record.request.executionTree?.rootObjectiveId === query.rootObjectiveId - ) - .filter((record) => !query.nodeId || record.request.executionTree?.nodeId === query.nodeId) - .filter( - (record) => - !query.parentNodeId || record.request.executionTree?.parentNodeId === query.parentNodeId - ) - .filter((record) => !states || states.has(record.state)) - .sort((left, right) => Date.parse(right.updatedAt) - Date.parse(left.updatedAt)) - .slice(0, query.limit ?? 100); + await this.prepareParent(); + return withFileLock(this.filePath, async () => { + const states = query.states ? new Set(query.states) : undefined; + return [...this.materialize(await this.readSnapshots()).values()] + .filter((record) => !query.workspaceId || record.request.workspaceId === query.workspaceId) + .filter((record) => !query.taskId || record.request.taskId === query.taskId) + .filter((record) => !query.rootTaskId || record.request.rootTaskId === query.rootTaskId) + .filter((record) => !query.provider || record.request.provider === query.provider) + .filter((record) => !query.hostId || record.request.hostId === query.hostId) + .filter( + (record) => !query.workflowRunId || record.request.workflowRunId === query.workflowRunId + ) + .filter( + (record) => + !query.workflowStepId || record.request.workflowStepId === query.workflowStepId + ) + .filter( + (record) => + !query.rootReservationId || record.request.rootReservationId === query.rootReservationId + ) + .filter( + (record) => + !query.rootObjectiveId || + record.request.executionTree?.rootObjectiveId === query.rootObjectiveId + ) + .filter((record) => !query.nodeId || record.request.executionTree?.nodeId === query.nodeId) + .filter( + (record) => + !query.parentNodeId || record.request.executionTree?.parentNodeId === query.parentNodeId + ) + .filter((record) => !states || states.has(record.state)) + .sort((left, right) => Date.parse(right.updatedAt) - Date.parse(left.updatedAt)) + .slice(0, query.limit ?? 100); + }); } async getQueueEntry(id: string): Promise { - return (await this.readQueueEntries()).get(id) ?? null; + await this.prepareParent(); + return withFileLock(this.filePath, async () => (await this.readQueueEntries()).get(id) ?? null); } async listQueue(query: AdmissionQueueListQuery): Promise { - const states = query.states ? new Set(query.states) : undefined; - return [...(await this.readQueueEntries()).values()] - .filter((entry) => !query.workspaceId || entry.request.workspaceId === query.workspaceId) - .filter((entry) => !query.taskId || entry.request.taskId === query.taskId) - .filter((entry) => !states || states.has(entry.state)) - .filter((entry) => !query.withSelectionEvidence || Boolean(entry.selectionEvidence)) - .filter( - (entry) => - !query.eligibleAt || Date.parse(entry.availableAt) <= Date.parse(query.eligibleAt) - ) - .sort((left, right) => - query.order === 'updated-desc' - ? Date.parse(right.updatedAt) - Date.parse(left.updatedAt) || - right.enqueueSequence - left.enqueueSequence || - left.id.localeCompare(right.id) - : left.enqueueSequence - right.enqueueSequence || left.id.localeCompare(right.id) - ) - .slice(0, query.limit ?? 100); + await this.prepareParent(); + return withFileLock(this.filePath, async () => { + const states = query.states ? new Set(query.states) : undefined; + return [...(await this.readQueueEntries()).values()] + .filter((entry) => !query.workspaceId || entry.request.workspaceId === query.workspaceId) + .filter((entry) => !query.taskId || entry.request.taskId === query.taskId) + .filter((entry) => !states || states.has(entry.state)) + .filter((entry) => !query.withSelectionEvidence || Boolean(entry.selectionEvidence)) + .filter( + (entry) => + !query.eligibleAt || Date.parse(entry.availableAt) <= Date.parse(query.eligibleAt) + ) + .sort((left, right) => + query.order === 'updated-desc' + ? Date.parse(right.updatedAt) - Date.parse(left.updatedAt) || + right.enqueueSequence - left.enqueueSequence || + left.id.localeCompare(right.id) + : left.enqueueSequence - right.enqueueSequence || left.id.localeCompare(right.id) + ) + .slice(0, query.limit ?? 100); + }); } async compareAndSetQueue(