From 4eb27427f809b565202f0b1bfa8129073f3fe5bd Mon Sep 17 00:00:00 2001 From: whyour Date: Sun, 13 Sep 2026 13:44:44 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20develop=20=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E9=94=81=E3=80=81=E8=BF=90=E8=A1=8C=E6=81=A2=E5=A4=8D?= =?UTF-8?q?=E4=B8=8E=20Alpine=20=E7=BC=93=E5=AD=98=E5=B9=B6=E5=8F=91=20(#3?= =?UTF-8?q?070)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: repair develop scheduler locks and runtime recovery * test: wait for child reaping after verified termination * fix: abort cron mutations when scheduler deletion fails --- back/config/util.ts | 30 +++++++-- back/schedule/api.ts | 16 +++-- back/schedule/client.ts | 6 +- back/services/cron.ts | 34 ++-------- back/shared/schedulerMutationLock.ts | 4 +- shell/node_path_cache.sh | 22 +++++-- test/back/grpc-cron-response.test.cjs | 60 ++++++++++++++++++ test/back/manual-stop-claim.test.cjs | 5 ++ test/back/process-exit-state.test.cjs | 81 ++++++++++++++++++++++++ test/back/scheduler-file-lock.test.cjs | 51 +++++++++++++++ test/back/scheduler-mutation.test.cjs | 38 +++++++++++ test/back/scheduler-rpc-timeout.test.cjs | 49 ++++++++++++++ 12 files changed, 350 insertions(+), 46 deletions(-) create mode 100644 test/back/grpc-cron-response.test.cjs create mode 100644 test/back/process-exit-state.test.cjs create mode 100644 test/back/scheduler-file-lock.test.cjs create mode 100644 test/back/scheduler-rpc-timeout.test.cjs diff --git a/back/config/util.ts b/back/config/util.ts index e6390ea9..dbd2f2a1 100644 --- a/back/config/util.ts +++ b/back/config/util.ts @@ -510,23 +510,45 @@ export async function killTask(pid: number, waitForExit = false) { } }; for (const target of pids) signal(target, 'SIGTERM'); - const alive = (target: number) => { + const alive = async (target: number) => { try { process.kill(target, 0); - return true; } catch (error: any) { if (error.code === 'ESRCH') return false; throw error; } + if (process.platform === 'linux') { + try { + const stat = await fs.readFile(`/proc/${target}/stat`, 'utf8'); + // The command field may contain spaces and parentheses. Zombies have + // exited even while their parent has not reaped the PID yet. + const state = stat.slice(stat.lastIndexOf(')') + 2).split(' ')[0]; + if (['Z', 'X', 'x'].includes(state)) return false; + } catch { + // /proc may be unavailable, or the process may have just exited. + // Retain the portable signal probe rather than assuming it is dead. + try { + process.kill(target, 0); + } catch (error: any) { + if (error.code === 'ESRCH') return false; + throw error; + } + } + } + return true; }; const wait = async () => { const deadline = Date.now() + 1000; - while (pids.some(alive) && Date.now() < deadline) { + let remaining = pids; + while (true) { + const states = await Promise.all(remaining.map(alive)); + remaining = remaining.filter((_, index) => states[index]); + if (!remaining.length || Date.now() >= deadline) return remaining; await new Promise((resolve) => setTimeout(resolve, 20)); } - return pids.filter(alive); }; let remaining = await wait(); + if (!remaining.length) return; for (const target of remaining) signal(target, 'SIGKILL'); remaining = await wait(); if (remaining.length) diff --git a/back/schedule/api.ts b/back/schedule/api.ts index 81747958..08f838ee 100644 --- a/back/schedule/api.ts +++ b/back/schedule/api.ts @@ -38,6 +38,7 @@ import { RunCronsRequest, } from '../protos/api'; import { NotificationInfo } from '../data/notify'; +import { Model } from 'sequelize'; Container.set('logger', LoggerInstance); @@ -247,13 +248,16 @@ export const systemNotify = async ( const normalizeCronData = (data: CronItem | null): CronItem | undefined => { if (!data) return undefined; + // create() returns a Sequelize instance; spreading it omits attribute getters. + const cron = data instanceof Model ? (data.get({ plain: true }) as CronItem) : data; return { - ...data, - sub_id: data.sub_id ?? undefined, - extra_schedules: data.extra_schedules ?? [], - pid: data.pid ?? undefined, - task_before: data.task_before ?? undefined, - task_after: data.task_after ?? undefined, + ...cron, + labels: cron.labels ?? [], + sub_id: cron.sub_id ?? undefined, + extra_schedules: cron.extra_schedules ?? [], + pid: cron.pid ?? undefined, + task_before: cron.task_before ?? undefined, + task_after: cron.task_after ?? undefined, }; }; diff --git a/back/schedule/client.ts b/back/schedule/client.ts index 3802288e..495ed12b 100644 --- a/back/schedule/client.ts +++ b/back/schedule/client.ts @@ -74,7 +74,9 @@ class Client { { deadline: Date.now() + 5000 }, (err, res) => { if (err) { - if (err.code === status.UNAVAILABLE) { + if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) { + // A timed-out write may already have reached the scheduler. + // Reconcile its state from the DB instead of replaying the RPC. this.readiness.invalidate(); Object.assign(err, { status: 503 }); } @@ -91,7 +93,7 @@ class Client { return new Promise((resolve, reject) => { this.client.delCron({ ids: request }, new Metadata(), { deadline: Date.now() + 5000 }, (err, res) => { if (err) { - if (err.code === status.UNAVAILABLE) { + if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) { this.readiness.invalidate(); Object.assign(err, { status: 503 }); } diff --git a/back/services/cron.ts b/back/services/cron.ts index a20f0966..afc4289f 100644 --- a/back/services/cron.ts +++ b/back/services/cron.ts @@ -154,20 +154,14 @@ export default class CronService { const tab = new Crontab({ ...doc, ...payload }); tab.saved = false; tab.log_name = await this.getLogName(tab); - const newDoc = await this.updateDb(tab); - if (doc.isDisabled === 1 || isDemoEnv()) { - return newDoc; + return await this.updateDb(tab); } - try { - await cronClient.delCron([String(newDoc.id)]); - } catch (error: any) { - this.logger.warn( - '[crontab] Failed to unregister cron job in scheduler:', - error?.message || error, - ); - } + // Keep the DB snapshot unchanged if deletion has an uncertain outcome. + // Recovery uses that snapshot after this mutation releases its lock. + await cronClient.delCron([String(doc.id)]); + const newDoc = await this.updateDb(tab); if (this.shouldUseCronClient(newDoc)) { try { @@ -305,15 +299,8 @@ export default class CronService { public async remove(ids: number[]) { return withSchedulerMutation(async () => { + await cronClient.delCron(ids.map(String)); await CrontabModel.destroy({ where: { id: ids } }); - try { - await cronClient.delCron(ids.map(String)); - } catch (error: any) { - this.logger.warn( - '[crontab] Failed to unregister cron job in scheduler:', - error?.message || error, - ); - } await this.setCrontab(); }); } @@ -853,15 +840,8 @@ export default class CronService { public async disabled(ids: number[]) { return withSchedulerMutation(async () => { + await cronClient.delCron(ids.map(String)); await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } }); - try { - await cronClient.delCron(ids.map(String)); - } catch (error: any) { - this.logger.warn( - '[crontab] Failed to unregister cron job in scheduler:', - error?.message || error, - ); - } await this.setCrontab(); }); } diff --git a/back/shared/schedulerMutationLock.ts b/back/shared/schedulerMutationLock.ts index afa2d4e0..a31c7fa3 100644 --- a/back/shared/schedulerMutationLock.ts +++ b/back/shared/schedulerMutationLock.ts @@ -9,7 +9,9 @@ export async function withSchedulerMutation( ): Promise { let release: () => Promise; try { - release = await lockfile.lock(config.crontabFile, { + // proper-lockfile indexes held locks by target, not lockfilePath. Keep + // this identity separate from nested writeFileWithLock(crontabFile). + release = await lockfile.lock(`${config.crontabFile}.scheduler`, { realpath: false, lockfilePath: `${config.crontabFile}.scheduler.lock`, stale: 30000, diff --git a/shell/node_path_cache.sh b/shell/node_path_cache.sh index 69a0691f..23363227 100644 --- a/shell/node_path_cache.sh +++ b/shell/node_path_cache.sh @@ -71,13 +71,23 @@ ql_refresh_node_global_path() ( if type -P flock &>/dev/null && mkdir -p -- "$dir_tmp" 2>/dev/null; then if [[ ! -L "$lock" && ( ! -e "$lock" || ( -f "$lock" && -O "$lock" ) ) ]] && \ { exec 9>> "$lock"; } 2>/dev/null; then - # Bound the wait; absent/unsupported flock or contention falls back to - # independent discovery. Never remove the lock file while waiters exist. - if flock -w 2 9 2>/dev/null; then - if ql_read_node_path_cache "$cache" "$key"; then - return 0 + # BusyBox flock (Alpine) has no -w. Both implementations support -n; + # bound contention retries and fall back immediately on other failures. + # Never remove the lock file while waiters exist. + local lock_attempt=0 lock_status + while :; do + if flock -n 9 2>/dev/null; then + if ql_read_node_path_cache "$cache" "$key"; then + return 0 + fi + break + else + lock_status=$? fi - fi + [[ "$lock_status" == 1 && "$lock_attempt" -lt 20 ]] || break + lock_attempt=$((lock_attempt + 1)) + sleep 0.1 + done fi fi diff --git a/test/back/grpc-cron-response.test.cjs b/test/back/grpc-cron-response.test.cjs new file mode 100644 index 00000000..e574a303 --- /dev/null +++ b/test/back/grpc-cron-response.test.cjs @@ -0,0 +1,60 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const ts = require('typescript'); +const { Sequelize, DataTypes, Model } = require('sequelize'); +const load = require('../helpers/load-security-module.cjs'); +const { CronResponse } = load('back/protos/api.ts'); +const source = fs.readFileSync('back/schedule/api.ts', 'utf8'); +const start = source.indexOf('const normalizeCronData ='); +const end = source.indexOf('export const getCronDetail', start); +const code = ts.transpileModule(source.slice(start, end), { + compilerOptions: { target: ts.ScriptTarget.ES2020 }, +}).outputText; +const normalize = new Function('Model', `${code}\nreturn normalizeCronData;`)( + Model, +); + +test('gRPC cron response preserves Sequelize model fields through protobuf serialization', async (t) => { + const db = new Sequelize({ + dialect: 'sqlite', + storage: ':memory:', + logging: false, + }); + t.after(() => db.close()); + const Cron = db.define('Cron', { + name: DataTypes.STRING, + command: DataTypes.STRING, + schedule: DataTypes.STRING, + labels: DataTypes.JSON, + extra_schedules: DataTypes.JSON, + }); + const model = Cron.build({ + id: 7, + name: 'smoke', + command: 'true', + schedule: '* * * * *', + labels: ['test'], + }); + const decoded = CronResponse.decode( + CronResponse.encode({ code: 200, data: normalize(model) }).finish(), + ); + assert.equal(decoded.data.id, 7); + assert.equal(decoded.data.name, 'smoke'); + assert.equal(decoded.data.command, 'true'); + assert.deepEqual(decoded.data.labels, ['test']); + assert.deepEqual(decoded.data.extra_schedules, []); +}); + +test('legacy plain cron rows normalize absent repeated fields', () => { + const decoded = CronResponse.decode( + CronResponse.encode({ + code: 200, + data: normalize({ id: 8, labels: null, extra_schedules: null }), + }).finish(), + ); + assert.equal(decoded.data.id, 8); + assert.deepEqual(decoded.data.labels, []); + assert.deepEqual(decoded.data.extra_schedules, []); + assert.equal(normalize(null), undefined); +}); diff --git a/test/back/manual-stop-claim.test.cjs b/test/back/manual-stop-claim.test.cjs index 91866276..8597c236 100644 --- a/test/back/manual-stop-claim.test.cjs +++ b/test/back/manual-stop-claim.test.cjs @@ -4,6 +4,7 @@ const fs = require('node:fs/promises'); const os = require('node:os'); const path = require('node:path'); const { spawn } = require('node:child_process'); +const { once } = require('node:events'); const { Sequelize, DataTypes } = require('sequelize'); const load = require('../helpers/load-security-module.cjs'); const { killTask } = require('../../back/config/util'); @@ -147,8 +148,12 @@ test( if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); }); + const exited = once(child, 'exit'); await new Promise((resolve) => child.stdout.once('data', resolve)); await killTask(child.pid, true); + // Linux may report an exited zombie before Node reaps our child. Wait for + // that separate event before requiring the PID to disappear. + await exited; assert.throws(() => process.kill(child.pid, 0), { code: 'ESRCH' }); assert.equal(child.signalCode, 'SIGKILL'); }, diff --git a/test/back/process-exit-state.test.cjs b/test/back/process-exit-state.test.cjs new file mode 100644 index 00000000..33173932 --- /dev/null +++ b/test/back/process-exit-state.test.cjs @@ -0,0 +1,81 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const ts = require('typescript'); + +// Isolate OS probes while exercising the production termination algorithm. +function terminationFixture(readStat, signalProbe = () => {}) { + const source = fs.readFileSync('back/config/util.ts', 'utf8'); + const start = source.indexOf('export async function killTask('); + const end = source.indexOf('export async function getPid(', start); + const code = ts.transpileModule(source.slice(start, end), { + compilerOptions: { + target: ts.ScriptTarget.ES2020, + module: ts.ModuleKind.CommonJS, + }, + }).outputText; + const signals = []; + const exports = {}; + new Function('exports', 'process', 'psTree', 'fs', 'setTimeout', code)( + exports, + { + platform: 'linux', + kill: (pid, signal) => { + signals.push([pid, signal]); + signalProbe(pid, signal); + }, + }, + async () => [], + { readFile: readStat }, + setTimeout, + ); + return { killTask: exports.killTask, signals }; +} + +for (const state of ['Z', 'X', 'x']) { + test(`termination accepts Linux exited state ${state} before PID reaping`, async () => { + const { killTask, signals } = terminationFixture( + async () => `42 (name with ) parentheses) ${state} 1 0 0`, + ); + await killTask(42, true); + assert.equal( + signals.some(([, signal]) => signal === 'SIGKILL'), + false, + ); + }); +} + +test('termination keeps waiting while the process is running', async () => { + let reads = 0; + const { killTask } = terminationFixture( + async () => `42 (worker) ${++reads === 1 ? 'S' : 'Z'} 1 0 0`, + ); + await killTask(42, true); + assert.equal(reads, 2); +}); + +test('a process disappearing during the procfs read is accepted', async () => { + let probes = 0; + const { killTask } = terminationFixture( + async () => { + throw Object.assign(Error('gone'), { code: 'ENOENT' }); + }, + (_pid, signal) => { + if (signal === 0 && ++probes > 1) + throw Object.assign(Error('gone'), { code: 'ESRCH' }); + }, + ); + await killTask(42, true); + assert.equal(probes, 2); +}); + +test('permission failures are not reported as successful termination', async () => { + const denied = Object.assign(Error('denied'), { code: 'EPERM' }); + const { killTask } = terminationFixture( + async () => '', + () => { + throw denied; + }, + ); + await assert.rejects(killTask(42, true), (error) => error === denied); +}); diff --git a/test/back/scheduler-file-lock.test.cjs b/test/back/scheduler-file-lock.test.cjs new file mode 100644 index 00000000..b469478b --- /dev/null +++ b/test/back/scheduler-file-lock.test.cjs @@ -0,0 +1,51 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs/promises'); +const os = require('node:os'); +const path = require('node:path'); +const load = require('../helpers/load-security-module.cjs'); + +for (const failOperation of [false, true]) { + test(`scheduler lock survives nested file writes and releases after ${ + failOperation ? 'failure' : 'success' + }`, async (t) => { + // Match production's canonical /ql path even when macOS temp is a symlink. + const root = await fs.realpath( + await fs.mkdtemp(path.join(os.tmpdir(), 'ql-nested-lock-')), + ); + t.after(() => fs.rm(root, { recursive: true, force: true })); + const config = { crontabFile: path.join(root, 'crontab.list') }; + await fs.writeFile(config.crontabFile, ''); + const { withSchedulerMutation } = load( + 'back/shared/schedulerMutationLock.ts', + { + '../config': config, + }, + ); + const { writeFileWithLock } = load('back/shared/utils.ts', { + '../config/util': { + fileExist: async (file) => + fs.access(file).then( + () => true, + () => false, + ), + }, + }); + const failure = new Error('registration failed'); + const operation = withSchedulerMutation(async () => { + await writeFileWithLock(config.crontabFile, 'first'); + await writeFileWithLock(config.crontabFile, 'second'); + if (failOperation) throw failure; + }); + if (failOperation) + await assert.rejects(operation, (error) => error === failure); + else await operation; + await assert.rejects(fs.stat(`${config.crontabFile}.scheduler.lock`), { + code: 'ENOENT', + }); + await withSchedulerMutation(() => + writeFileWithLock(config.crontabFile, 'recovered'), + ); + assert.equal(await fs.readFile(config.crontabFile, 'utf8'), 'recovered'); + }); +} diff --git a/test/back/scheduler-mutation.test.cjs b/test/back/scheduler-mutation.test.cjs index e975e913..36223c2f 100644 --- a/test/back/scheduler-mutation.test.cjs +++ b/test/back/scheduler-mutation.test.cjs @@ -7,6 +7,7 @@ const { spawn } = require('node:child_process'); const { once } = require('node:events'); const { setTimeout: delay } = require('node:timers/promises'); const { Sequelize, DataTypes } = require('sequelize'); +const { status } = require('@grpc/grpc-js'); const load = require('../helpers/load-security-module.cjs'); function gate() { @@ -247,6 +248,43 @@ test( }, ); +for (const operation of ['update', 'remove', 'disabled']) { + for (const code of [status.DEADLINE_EXCEEDED, status.UNAVAILABLE]) { + for (const applied of [false, true]) { + test(`${operation} aborts on RPC ${code} with deletion applied=${applied} and reconciles the original DB`, async (t) => { + const { service, crons, client, jobs } = await fixture(t); + await crons.create({ + id: 1, command: 'old', schedule: '* * * * *', isDisabled: 0, + }); + jobs.set('1', 'old'); + const before = await crons.findAll({ raw: true }); + const failure = Object.assign(Error('uncertain deletion'), { code, status: 503 }); + let deletes = 0; + client.delCron = async (ids) => { + deletes++; + if (applied) ids.forEach((id) => jobs.delete(id)); + throw failure; + }; + const add = client.addCron; + client.addCron = async () => assert.fail('must not register after failed deletion'); + service.setCrontab = async () => assert.fail('must not publish after failed deletion'); + await assert.rejects( + operation === 'update' + ? service.update({ id: 1, command: 'new' }) + : service[operation]([1]), + (error) => error === failure && error.status === 503, + ); + assert.equal(deletes, 1, 'must not replay an uncertain deletion'); + assert.deepEqual(await crons.findAll({ raw: true }), before); + client.addCron = add; + service.setCrontab = async () => {}; + await service.autosave_crontab(true); + assert.deepEqual([...jobs.entries()], [['1', 'old']]); + }); + } + } +} + test('configuration rollback cannot restore an obsolete manual queue token', async (t) => { const { service, crons, client } = await fixture(t); await crons.create({ diff --git a/test/back/scheduler-rpc-timeout.test.cjs b/test/back/scheduler-rpc-timeout.test.cjs new file mode 100644 index 00000000..e76d01aa --- /dev/null +++ b/test/back/scheduler-rpc-timeout.test.cjs @@ -0,0 +1,49 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const grpc = require('@grpc/grpc-js'); +const load = require('../helpers/load-security-module.cjs'); + +for (const method of ['addCron', 'delCron']) { + test(`${method} invalidates an uncertain timed-out write without replaying it`, async () => { + let writes = 0; + let invalidations = 0; + const timeout = Object.assign(new Error('response deadline exceeded'), { + code: grpc.status.DEADLINE_EXCEEDED, + }); + const fake = { + waitForReady: (_deadline, callback) => callback(), + [method]: (_request, _metadata, _options, callback) => { + // The server may have applied the write before the response was lost. + writes++; + callback(timeout); + }, + }; + const client = load('back/schedule/client.ts', { + '../protos/cron': { + CronClient: class { + constructor() { + return fake; + } + }, + }, + '../config': { grpcPort: 5500 }, + '../config/grpcCerts': { + getGrpcCerts: () => ({ + caCert: 'ca', + clientKey: 'key', + clientCert: 'cert', + }), + }, + '@grpc/grpc-js': { ...grpc, credentials: { createSsl: () => ({}) } }, + }).default; + client.readiness.invalidate = () => invalidations++; + await assert.rejects(client[method]([]), (error) => error === timeout); + assert.equal( + invalidations, + 1, + 'reconcile from the DB after an uncertain RPC result', + ); + assert.equal(timeout.status, 503); + assert.equal(writes, 1, 'do not replay an uncertain mutation'); + }); +}