mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: fall back to hot code queue reads

Browse Source
This commit is contained in:
Codex
2026-05-17 08:34:45 +00:00
parent c93039bf95
commit 48db823e52
2 changed files with 116 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+10 -4
View File
@@ -1634,10 +1634,16 @@ function rememberHotTask(task: QueueTask): QueueTask {
async function findTaskForRead(taskId: string): Promise<QueueTask | null> {
const hotTask = findTask(taskId);
if (!databaseReady) return hotTask;
const databaseTask = await loadTaskFromDatabase(taskId);
if (databaseTask === null) return hotTask;
if (hotTask !== null && (timestampMs(hotTask.updatedAt) ?? 0) > (timestampMs(databaseTask.updatedAt) ?? 0)) return hotTask;
return databaseTask;
try {
const databaseTask = await loadTaskFromDatabase(taskId);
if (databaseTask === null) return hotTask;
if (hotTask !== null && (timestampMs(hotTask.updatedAt) ?? 0) > (timestampMs(databaseTask.updatedAt) ?? 0)) return hotTask;
return databaseTask;
} catch (error) {
databaseLastError = databaseErrorMessage(error);
logger("warn", "read_database_fallback", { taskId, error: errorToJson(error) });
return hotTask;
}
}
async function findTaskForMutation(taskId: string): Promise<QueueTask | null> {
src/components/microservices/code-queue/src/queue-api.ts
+106 -52
View File
@@ -82,22 +82,43 @@ function mergeReadTaskByFreshness(target: Map<string, QueueTask>, task: QueueTas
if (existing === undefined || memoryTaskIsNewerThanDatabase(task, existing)) target.set(task.id, task);
}
function rememberReadDatabaseError(error: unknown): void {
const message = error instanceof Error ? error.message : String(error);
console.warn(JSON.stringify({
ts: new Date().toISOString(),
service: "code-queue",
level: "warn",
message: "read_database_fallback",
data: { error: message },
}));
}
async function loadFreshMemoryTasksForRead(): Promise<QueueTask[]> {
const memoryTasks = ctx().tasks();
if (!ctx().databaseReady() || memoryTasks.length === 0) return memoryTasks;
const databaseTasks = await ctx().loadTasksFromDatabaseByIds(memoryTasks.map((task) => task.id));
const byId = new Map(databaseTasks.map((task) => [task.id, task]));
for (const task of memoryTasks) mergeReadTaskByFreshness(byId, task);
return Array.from(byId.values());
try {
const databaseTasks = await ctx().loadTasksFromDatabaseByIds(memoryTasks.map((task) => task.id));
const byId = new Map(databaseTasks.map((task) => [task.id, task]));
for (const task of memoryTasks) mergeReadTaskByFreshness(byId, task);
return Array.from(byId.values());
} catch (error) {
rememberReadDatabaseError(error);
return memoryTasks;
}
}
async function loadTaskForReadByFreshness(taskId: string): Promise<QueueTask | null> {
const memoryTask = ctx().tasks().find((task) => task.id === taskId) ?? null;
if (!ctx().databaseReady()) return memoryTask;
const databaseTask = await ctx().loadTaskFromDatabase(taskId);
if (databaseTask === null) return memoryTask;
if (memoryTask !== null && memoryTaskIsNewerThanDatabase(memoryTask, databaseTask)) return memoryTask;
return databaseTask;
try {
const databaseTask = await ctx().loadTaskFromDatabase(taskId);
if (databaseTask === null) return memoryTask;
if (memoryTask !== null && memoryTaskIsNewerThanDatabase(memoryTask, databaseTask)) return memoryTask;
return databaseTask;
} catch (error) {
rememberReadDatabaseError(error);
return memoryTask;
}
}
function queuedStatusPayload(task: QueueTask, tasks?: QueueTask[]): { queuedReason: QueuedStatusReason | null; queuedReasonLabel: string | null } {
@@ -472,13 +493,18 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
async function loadAllTasksForRead(): Promise<QueueTask[]> {
if (!ctx().databaseReady()) return ctx().tasks();
const tasks = await ctx().loadTasksFromDatabase("all");
const byId = new Map(tasks.map((task) => [task.id, task]));
for (const active of ctx().tasks()) {
mergeReadTaskByFreshness(byId, active);
try {
const tasks = await ctx().loadTasksFromDatabase("all");
const byId = new Map(tasks.map((task) => [task.id, task]));
for (const active of ctx().tasks()) {
mergeReadTaskByFreshness(byId, active);
}
ctx().runGarbageCollection();
return Array.from(byId.values()).sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id));
} catch (error) {
rememberReadDatabaseError(error);
return ctx().tasks();
}
ctx().runGarbageCollection();
return Array.from(byId.values()).sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id));
}
async function queueSummaryForResponse(includeDevReady = true, tasks?: QueueTask[], queueRecords?: QueueRecord[]): Promise<JsonValue> {
@@ -487,40 +513,53 @@ async function queueSummaryForResponse(includeDevReady = true, tasks?: QueueTask
}
async function queueSummaryForHealth(includeDevReady = true): Promise<JsonValue> {
const queueRecords = ctx().databaseReady() ? await ctx().loadQueuesFromDatabase() : ctx().queues();
let queueRecords = ctx().queues();
if (ctx().databaseReady()) {
try {
queueRecords = await ctx().loadQueuesFromDatabase();
} catch (error) {
rememberReadDatabaseError(error);
}
}
const hotTasks = await loadFreshMemoryTasksForRead();
const summary = queueSummary(includeDevReady, hotTasks, queueRecords) as Record<string, JsonValue>;
if (!ctx().databaseReady()) return summary;
const aggregateRows = await ctx().sql<QueueSummaryAggregateRow[]>`
SELECT
(SELECT COUNT(*) FROM unidesk_code_queue_tasks) AS total,
COALESCE((
SELECT jsonb_object_agg(status, count)
FROM (
SELECT status, COUNT(*) AS count
FROM unidesk_code_queue_tasks
GROUP BY status
) AS status_counts
), '{}'::jsonb) AS status_counts,
COALESCE((
SELECT jsonb_agg(jsonb_build_object('queueId', queue_id, 'status', status, 'count', count))
FROM (
SELECT queue_id, status, COUNT(*) AS count
FROM unidesk_code_queue_tasks
GROUP BY queue_id, status
) AS queue_status_counts
), '[]'::jsonb) AS queue_status_counts,
COALESCE((
SELECT jsonb_agg(jsonb_build_object('queueId', queue_id, 'count', count))
FROM (
SELECT queue_id, COUNT(*) AS count
FROM unidesk_code_queue_tasks
WHERE status IN ('succeeded', 'failed', 'canceled')
AND read_at IS NULL
GROUP BY queue_id
) AS unread_counts
), '[]'::jsonb) AS unread_counts
`;
let aggregateRows: QueueSummaryAggregateRow[];
try {
aggregateRows = await ctx().sql<QueueSummaryAggregateRow[]>`
SELECT
(SELECT COUNT(*) FROM unidesk_code_queue_tasks) AS total,
COALESCE((
SELECT jsonb_object_agg(status, count)
FROM (
SELECT status, COUNT(*) AS count
FROM unidesk_code_queue_tasks
GROUP BY status
) AS status_counts
), '{}'::jsonb) AS status_counts,
COALESCE((
SELECT jsonb_agg(jsonb_build_object('queueId', queue_id, 'status', status, 'count', count))
FROM (
SELECT queue_id, status, COUNT(*) AS count
FROM unidesk_code_queue_tasks
GROUP BY queue_id, status
) AS queue_status_counts
), '[]'::jsonb) AS queue_status_counts,
COALESCE((
SELECT jsonb_agg(jsonb_build_object('queueId', queue_id, 'count', count))
FROM (
SELECT queue_id, COUNT(*) AS count
FROM unidesk_code_queue_tasks
WHERE status IN ('succeeded', 'failed', 'canceled')
AND read_at IS NULL
GROUP BY queue_id
) AS unread_counts
), '[]'::jsonb) AS unread_counts
`;
} catch (error) {
rememberReadDatabaseError(error);
return summary as JsonValue;
}
const aggregate = aggregateRows[0] ?? { total: ctx().tasks().length, status_counts: {}, queue_status_counts: [], unread_counts: [] };
const counts: Record<string, number> = {};
for (const [status, count] of Object.entries(aggregate.status_counts ?? {})) counts[status] = Number(count);
@@ -1065,12 +1104,21 @@ async function databaseTasksOverviewResponse(url: URL): Promise<Response | null>
const includeStatistics = url.searchParams.get("stats") !== "0";
const priorityLimit = parseNamedLimit(url, "priorityLimit", Math.max(100, limit));
const beforeId = url.searchParams.get("beforeId");
const [queue, statistics, page, priorityIds] = await Promise.all([
queueSummaryForResponse(false),
includeStatistics ? databaseTaskStatisticsSummary(queueId, url) : Promise.resolve(null),
databasePageTaskIds(queueId, beforeId, limit),
includeActive ? databasePriorityTaskIds(queueId, priorityLimit) : Promise.resolve([]),
]);
let queue: JsonValue;
let statistics: JsonValue | null;
let page: { ids: string[]; total: number; hasMore: boolean; nextBeforeId: string | null };
let priorityIds: string[];
try {
[queue, statistics, page, priorityIds] = await Promise.all([
queueSummaryForResponse(false),
includeStatistics ? databaseTaskStatisticsSummary(queueId, url) : Promise.resolve(null),
databasePageTaskIds(queueId, beforeId, limit),
includeActive ? databasePriorityTaskIds(queueId, priorityLimit) : Promise.resolve([]),
]);
} catch (error) {
rememberReadDatabaseError(error);
return null;
}
const orderedIds: string[] = [];
const seenIds = new Set<string>();
for (const id of priorityIds) pushUniqueId(orderedIds, seenIds, id);
@@ -1079,7 +1127,13 @@ async function databaseTasksOverviewResponse(url: URL): Promise<Response | null>
if (!includeActive || !taskMatchesQueueFilter(task, queueId)) continue;
if (terminalTaskUnread(task) || activePriority(task) < 3) pushUniqueId(orderedIds, seenIds, task.id);
}
const loadedTasks = await ctx().loadTasksFromDatabaseByIds(orderedIds);
let loadedTasks: QueueTask[];
try {
loadedTasks = await ctx().loadTasksFromDatabaseByIds(orderedIds);
} catch (error) {
rememberReadDatabaseError(error);
return null;
}
const byId = new Map<string, QueueTask>();
for (const task of loadedTasks) byId.set(task.id, task);
for (const task of ctx().tasks()) {