fix: 修复 develop 调度锁、运行恢复与 Alpine 缓存并发 (#3070)

* 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

@@ -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)

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,
+    ...cron,
-    sub_id: data.sub_id ?? undefined,
+    labels: cron.labels ?? [],
-    extra_schedules: data.extra_schedules ?? [],
+    sub_id: cron.sub_id ?? undefined,
-    pid: data.pid ?? undefined,
+    extra_schedules: cron.extra_schedules ?? [],
-    task_before: data.task_before ?? undefined,
+    pid: cron.pid ?? undefined,
-    task_after: data.task_after ?? undefined,
+    task_before: cron.task_before ?? undefined,
+    task_after: cron.task_after ?? undefined,
   };
 };
 

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 });
           }

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 {
+      // Keep the DB snapshot unchanged if deletion has an uncertain outcome.
-        await cronClient.delCron([String(newDoc.id)]);
+      // Recovery uses that snapshot after this mutation releases its lock.
-      } catch (error: any) {
+      await cronClient.delCron([String(doc.id)]);
-        this.logger.warn(
+      const newDoc = await this.updateDb(tab);
-          '[crontab] Failed to unregister cron job in scheduler:',
-          error?.message || error,
-        );
-      }
 
       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();
     });
   }

back/shared/schedulerMutationLock.ts

@@ -9,7 +9,9 @@ export async function withSchedulerMutation<T>(
 ): Promise<T> {
   let release: () => Promise<void>;
   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,

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
+      # BusyBox flock (Alpine) has no -w. Both implementations support -n;
-      # independent discovery. Never remove the lock file while waiters exist.
+      # bound contention retries and fall back immediately on other failures.
-      if flock -w 2 9 2>/dev/null; then
+      # Never remove the lock file while waiters exist.
-        if ql_read_node_path_cache "$cache" "$key"; then
+      local lock_attempt=0 lock_status
-          return 0
+      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
 

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);
+});

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');
   },

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);
+});

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');
+  });
+}

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({

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');
+  });
+}