mirror of
https://github.com/BradGroux/veritas-kanban.git
synced 2026-10-06 19:57:50 +00:00
fix: serialize admission repository reads
This commit is contained in:
Brad Groux
parent
b261505409
commit
ef86d99a5e
2 changed files with 117 additions and 50 deletions
|
|
@ -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<ReturnType<typeof repository.list>> | 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<void>((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 = {
|
||||
|
|
|
|||
|
|
@ -273,65 +273,77 @@ export class FileAdmissionReservationRepository implements AdmissionReservationR
|
|||
}
|
||||
|
||||
async get(id: string): Promise<AdmissionReservation | null> {
|
||||
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<AdmissionReservation[]> {
|
||||
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<AdmissionQueueEntry | null> {
|
||||
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<AdmissionQueueEntry[]> {
|
||||
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(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue