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

Make Code Queue read role database startup read-only

Browse Source
This commit is contained in:
Codex
2026-05-17 08:14:09 +00:00
parent 57402f28c0
commit 0ed5c31509
1 changed files with 96 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+96 -84
View File
@@ -283,6 +283,10 @@ function serviceRoleAllowsScheduler(role: CodeQueueServiceRole): boolean {
return role === "combined" || role === "scheduler";
}
function serviceRoleReadOnly(role: CodeQueueServiceRole): boolean {
return role === "read";
}
function envList(name: string, fallback: string[]): string[] {
const raw = process.env[name];
const source = raw === undefined || raw.length === 0 ? fallback.join(",") : raw;
@@ -1158,6 +1162,7 @@ function runGarbageCollection(): void {
}
function scheduleDatabaseFlush(delayMs = config.databaseFlushIntervalMs): void {
if (serviceRoleReadOnly(config.serviceRole)) return;
if (!databaseReady || (dirtyDatabaseTaskIds.size === 0 && dirtyDatabaseQueueIds.size === 0) || shutdownRequested) return;
if (databaseFlushTimer !== null) return;
databaseFlushTimer = setTimeout(() => {
@@ -1924,6 +1929,11 @@ async function loadNextSeqFromDatabase(): Promise<number> {
}
async function flushDirtyTasksToDatabase(force = false): Promise<void> {
if (serviceRoleReadOnly(config.serviceRole)) {
dirtyDatabaseTaskIds.clear();
dirtyDatabaseQueueIds.clear();
return;
}
if (!databaseReady) return;
if (databaseFlushInFlight && !force) {
scheduleDatabaseFlush();
@@ -1964,89 +1974,91 @@ async function flushDirtyTasksToDatabase(force = false): Promise<void> {
async function initDatabasePersistence(): Promise<void> {
logger("info", "database_persistence_init_start", { databaseUrl: redactDatabaseUrl(config.databaseUrl) });
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_tasks (
id TEXT PRIMARY KEY,
queue_id TEXT NOT NULL DEFAULT 'default',
status TEXT NOT NULL,
provider_id TEXT NOT NULL DEFAULT 'main-server',
execution_mode TEXT NOT NULL DEFAULT 'default',
model TEXT NOT NULL,
cwd TEXT NOT NULL,
prompt TEXT NOT NULL,
base_prompt TEXT NOT NULL DEFAULT '',
reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
reference_injection JSONB,
reasoning_effort TEXT,
max_attempts INTEGER NOT NULL,
current_attempt INTEGER NOT NULL DEFAULT 0,
current_mode TEXT,
codex_thread_id TEXT,
active_turn_id TEXT,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
started_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
read_at TIMESTAMPTZ,
last_error TEXT,
last_judge JSONB,
output_count INTEGER NOT NULL DEFAULT 0,
event_count INTEGER NOT NULL DEFAULT 0,
attempt_count INTEGER NOT NULL DEFAULT 0,
last_output_seq BIGINT NOT NULL DEFAULT 0,
task_json JSONB NOT NULL
)
`;
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_queues (
id TEXT PRIMARY KEY,
name TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
)
`;
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_workdirs (
provider_id TEXT NOT NULL,
execution_mode TEXT NOT NULL DEFAULT 'default',
path TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (provider_id, execution_mode, path)
)
`;
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_notifications (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
dedup_key TEXT NOT NULL,
target TEXT NOT NULL,
message TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at TIMESTAMPTZ NOT NULL,
last_error TEXT,
sent_at TIMESTAMPTZ
)
`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS queue_id TEXT NOT NULL DEFAULT 'default'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS provider_id TEXT NOT NULL DEFAULT 'main-server'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS base_prompt TEXT NOT NULL DEFAULT ''`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_injection JSONB`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ`;
await sql`
UPDATE unidesk_code_queue_tasks
SET read_at = NULLIF(task_json->>'readAt', '')::timestamptz
WHERE read_at IS NULL
AND status IN ('succeeded', 'failed', 'canceled')
AND COALESCE(task_json->>'readAt', '') <> ''
AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T'
`;
await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
await sql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
if (!serviceRoleReadOnly(config.serviceRole)) {
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_tasks (
id TEXT PRIMARY KEY,
queue_id TEXT NOT NULL DEFAULT 'default',
status TEXT NOT NULL,
provider_id TEXT NOT NULL DEFAULT 'main-server',
execution_mode TEXT NOT NULL DEFAULT 'default',
model TEXT NOT NULL,
cwd TEXT NOT NULL,
prompt TEXT NOT NULL,
base_prompt TEXT NOT NULL DEFAULT '',
reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
reference_injection JSONB,
reasoning_effort TEXT,
max_attempts INTEGER NOT NULL,
current_attempt INTEGER NOT NULL DEFAULT 0,
current_mode TEXT,
codex_thread_id TEXT,
active_turn_id TEXT,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
started_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
read_at TIMESTAMPTZ,
last_error TEXT,
last_judge JSONB,
output_count INTEGER NOT NULL DEFAULT 0,
event_count INTEGER NOT NULL DEFAULT 0,
attempt_count INTEGER NOT NULL DEFAULT 0,
last_output_seq BIGINT NOT NULL DEFAULT 0,
task_json JSONB NOT NULL
)
`;
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_queues (
id TEXT PRIMARY KEY,
name TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
)
`;
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_workdirs (
provider_id TEXT NOT NULL,
execution_mode TEXT NOT NULL DEFAULT 'default',
path TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (provider_id, execution_mode, path)
)
`;
await sql`
CREATE TABLE IF NOT EXISTS unidesk_code_queue_notifications (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
dedup_key TEXT NOT NULL,
target TEXT NOT NULL,
message TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at TIMESTAMPTZ NOT NULL,
last_error TEXT,
sent_at TIMESTAMPTZ
)
`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS queue_id TEXT NOT NULL DEFAULT 'default'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS provider_id TEXT NOT NULL DEFAULT 'main-server'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS base_prompt TEXT NOT NULL DEFAULT ''`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_injection JSONB`;
await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ`;
await sql`
UPDATE unidesk_code_queue_tasks
SET read_at = NULLIF(task_json->>'readAt', '')::timestamptz
WHERE read_at IS NULL
AND status IN ('succeeded', 'failed', 'canceled')
AND COALESCE(task_json->>'readAt', '') <> ''
AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T'
`;
await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
await sql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
}
const countRows = await sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM unidesk_code_queue_tasks`;
const hotTasks = await loadTasksFromDatabase("hot");
@@ -2091,7 +2103,7 @@ async function initDatabasePersistence(): Promise<void> {
}
}
ensureDefaultWorkdirRecords();
await upsertWorkdirsToDatabase(sortedWorkdirRecords());
if (!serviceRoleReadOnly(config.serviceRole)) await upsertWorkdirsToDatabase(sortedWorkdirRecords());
databaseReady = true;
if (config.serviceRole === "combined" || config.serviceRole === "scheduler") scheduleStartupDatabaseMaintenance();
runGarbageCollection();