fix: 修复任务生命周期与调度就绪，优化执行和构建开销 (#3069)

* fix: harden task lifecycle and scheduler readiness

* fix: confine log writes to the configured log directory

* fix: verify complete build artifacts and untracked inputs

* fix: reconcile scheduler state and make stop win startup races

* fix: isolate cron generations and serialize scheduler recovery

back/shared/childProcess.ts

@@ -0,0 +1,78 @@
+import { ChildProcessWithoutNullStreams } from 'child_process';
+import { Readable } from 'stream';
+
+export interface ProcessResult {
+  code: number | null;
+  signal: NodeJS.Signals | null;
+  error?: Error;
+}
+
+export function asError(error: unknown): Error {
+  return error instanceof Error ? error : new Error(String(error));
+}
+
+/** Attach immediately after spawn, before awaiting database or user callbacks. */
+export function observeChildProcess(
+  child: ChildProcessWithoutNullStreams,
+  callbacks: {
+    onStart?: () => Promise<void>;
+    onStdout?: (message: string) => Promise<void>;
+    onStderr?: (message: string) => Promise<void>;
+  } = {},
+) {
+  let failure: Error | undefined;
+  const recordError = (error: unknown) => {
+    failure ??= asError(error);
+  };
+  const spawned = new Promise<void>((resolve, reject) => {
+    child.once('spawn', resolve);
+    // Keep the listener through close: errors can occur after a successful spawn.
+    child.on('error', (error) => {
+      recordError(error);
+      reject(error);
+    });
+  });
+  const closed = new Promise<ProcessResult>((resolve) => {
+    child.once('close', (code, signal) => resolve({ code, signal }));
+  });
+  const started = spawned.then(async () => {
+    await callbacks.onStart?.();
+    return child.pid;
+  });
+  // The caller can ask only for completion, without an unhandled start rejection.
+  const ready = started.catch(recordError);
+
+  const consume = async (
+    stream: Readable,
+    callback?: (message: string) => Promise<void>,
+  ) => {
+    // StringDecoder in Readable preserves UTF-8 characters split across chunks.
+    stream.setEncoding('utf8');
+    let callbackFailed = false;
+    try {
+      for await (const chunk of stream) {
+        await ready;
+        if (!callbackFailed && callback) {
+          try {
+            await callback(String(chunk));
+          } catch (error) {
+            recordError(error);
+            callbackFailed = true;
+          }
+        }
+        // Even if a log sink fails, drain the pipe so the child can finish.
+      }
+    } catch (error) {
+      recordError(error);
+    }
+  };
+  const output = Promise.all([
+    consume(child.stdout, callbacks.onStdout),
+    consume(child.stderr, callbacks.onStderr),
+  ]);
+  const completed = Promise.all([closed, ready, output]).then(([result]) => ({
+    ...result,
+    error: failure,
+  }));
+  return { started, completed };
+}

back/shared/logStreamManager.ts

@@ -1,5 +1,8 @@
 import { createWriteStream, WriteStream } from 'fs';
 import { EventEmitter } from 'events';
+import path from 'path';
+import config from '../config';
+import { resolveFileAccess } from './fileAccess';
 
 /**
  * Manages write streams for log files to improve performance by avoiding repeated file opens
@@ -8,83 +11,90 @@ export class LogStreamManager extends EventEmitter {
   private streams: Map<string, WriteStream> = new Map();
   private pendingWrites: Map<string, Promise<void>> = new Map();
 
-  /**
-   * Write data to a log file using a managed stream
-   * @param filePath - Absolute path to the log file
-   * @param data - Data to write to the log file
-   */
-  async write(filePath: string, data: string): Promise<void> {
-    // Wait for any pending writes to this file to complete
-    const pending = this.pendingWrites.get(filePath);
-    if (pending) {
-      await pending;
-    }
+  private closingStreams = new Map<string, Promise<void>>();
+  private closedStreams = new WeakSet<WriteStream>();
+  private streamErrors = new Map<string, Error>();
 
-    // Create a new promise for this write operation
-    const writePromise = new Promise<void>((resolve, reject) => {
-      let stream = this.streams.get(filePath);
-
-      if (!stream) {
-        // Create a new write stream if one doesn't exist
-        stream = createWriteStream(filePath, { flags: 'a' });
-        this.streams.set(filePath, stream);
-
-        // Handle stream errors
-        stream.on('error', (error) => {
-          this.emit('error', { filePath, error });
-          // Remove the stream from the map on error
-          this.streams.delete(filePath);
-          reject(error);
-        });
-      }
-
-      // Write the data
-      const canContinue = stream.write(data, 'utf8', (error) => {
-        if (error) {
-          reject(error);
-        } else {
-          resolve();
-        }
-      });
-
-      // Handle backpressure
-      if (!canContinue) {
-        stream.once('drain', () => {
-          // Stream is ready for more data
-        });
-      }
-    });
-
-    this.pendingWrites.set(filePath, writePromise);
-
-    try {
-      await writePromise;
-    } finally {
-      this.pendingWrites.delete(filePath);
-    }
+  constructor(private readonly logRoot = config.logPath) {
+    super();
   }
 
-  /**
-   * Close the stream for a specific file path
-   * @param filePath - Absolute path to the log file
-   */
-  async closeStream(filePath: string): Promise<void> {
-    // Wait for any pending writes to complete
-    const pending = this.pendingWrites.get(filePath);
-    if (pending) {
-      await pending.catch(() => {
-        // Ignore errors on pending writes during close
-      });
+  /** Register each write synchronously, so concurrent callers cannot lose the tail. */
+  async write(filePath: string, data: string): Promise<void> {
+    if (this.closingStreams.has(filePath)) {
+      throw new Error(`Log stream is closing: ${filePath}`);
     }
+    const previous = this.pendingWrites.get(filePath) || Promise.resolve();
+    const pending = previous.then(
+      () =>
+        new Promise<void>((resolve, reject) => {
+          const failure = this.streamErrors.get(filePath);
+          if (failure) return reject(failure);
+          let stream = this.streams.get(filePath);
+          if (!stream) {
+            // Validate only when opening: subsequent chunks reuse the same descriptor.
+            const root = path.resolve(this.logRoot);
+            const target = path.resolve(filePath);
+            if (
+              !target.startsWith(root + path.sep) ||
+              !resolveFileAccess(root, [target])
+            ) {
+              return reject(new Error('Log path is outside the log directory'));
+            }
+            stream = createWriteStream(target, { flags: 'a' });
+            this.streams.set(filePath, stream);
+            const current = stream;
+            stream.once('close', () => this.closedStreams.add(current));
+            stream.on('error', (error) => {
+              this.streamErrors.set(filePath, error);
+              // EventEmitter's unobserved "error" event would crash the caller.
+              if (this.listenerCount('error') > 0)
+                this.emit('error', { filePath, error });
+            });
+          }
+          stream.write(data, 'utf8', (error) =>
+            error ? reject(error) : resolve(),
+          );
+        }),
+    );
+    this.pendingWrites.set(filePath, pending);
+    // Keep the tail until close, including failures; never reopen a failed log mid-run.
+    return pending;
+  }
 
-    const stream = this.streams.get(filePath);
-    if (stream) {
-      return new Promise<void>((resolve) => {
-        stream.end(() => {
-          this.streams.delete(filePath);
-          resolve();
-        });
-      });
+  async closeStream(filePath: string): Promise<void> {
+    const closing = this.closingStreams.get(filePath);
+    if (closing) return closing;
+    const pending = this.pendingWrites.get(filePath);
+    const result = (async () => {
+      let failure: unknown;
+      try {
+        await pending;
+      } catch (error) {
+        failure = error;
+      }
+      const stream = this.streams.get(filePath);
+      try {
+        if (stream && !this.closedStreams.has(stream)) {
+          await new Promise<void>((resolve) => {
+            stream.once('close', resolve);
+            if (failure || stream.destroyed) stream.destroy();
+            else stream.end();
+          });
+        }
+        failure ||= this.streamErrors.get(filePath);
+        if (failure) throw failure;
+      } finally {
+        this.streams.delete(filePath);
+        this.pendingWrites.delete(filePath);
+        this.streamErrors.delete(filePath);
+      }
+    })();
+    this.closingStreams.set(filePath, result);
+    try {
+      await result;
+    } finally {
+      this.closingStreams.delete(filePath);
     }
   }
 
@@ -92,7 +102,11 @@ export class LogStreamManager extends EventEmitter {
    * Close all open streams
    */
   async closeAll(): Promise<void> {
-    const closePromises = Array.from(this.streams.keys()).map((filePath) =>
+    const paths = new Set([
+      ...this.streams.keys(),
+      ...this.pendingWrites.keys(),
+    ]);
+    const closePromises = Array.from(paths).map((filePath) =>
       this.closeStream(filePath),
     );
     await Promise.all(closePromises);

back/shared/runCron.ts

@@ -4,15 +4,13 @@ import Logger from '../loaders/logger';
 import { ICron } from '../protos/cron';
 import { CrontabModel, CrontabStatus } from '../data/cron';
 import { killTask } from '../config/util';
-import {
-  RunningInstanceModel,
-  InstanceStatus,
-} from '../data/runningInstance';
+import { RunningInstanceModel, InstanceStatus } from '../data/runningInstance';
 import dayjs from 'dayjs';
+import { observeChildProcess, asError } from './childProcess';
 
 export function runCron(cmd: string, cron: ICron): Promise<number | void> {
-  return taskLimit.runWithCronLimit(cron, () => {
-    return new Promise(async (resolve: any) => {
+  return taskLimit.runWithCronLimit(cron, async () => {
+    try {
       // Check if the cron is already running and stop it (only if multiple instances are not allowed)
       try {
         const existingCron = await CrontabModel.findOne({
@@ -38,7 +36,12 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
           const stoppedAt = dayjs().unix();
           await RunningInstanceModel.update(
             { status: InstanceStatus.stopped, finished_at: stoppedAt },
-            { where: { cron_id: Number(cron.id), status: InstanceStatus.running } },
+            {
+              where: {
+                cron_id: Number(cron.id),
+                status: InstanceStatus.running,
+              },
+            },
           );
           // Update the status to idle after killing
           await CrontabModel.update(
@@ -60,33 +63,37 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
       );
       const cp = spawn(cmd, { shell: '/bin/bash' });
 
-      cp.stderr.on('data', (data) => {
-        Logger.info(
-          '[schedule][执行任务失败] 命令: %s, 错误信息: %j',
-          cmd,
-          data.toString(),
-        );
+      const { completed } = observeChildProcess(cp, {
+        onStderr: async (message) => {
+          Logger.info(
+            '[schedule][任务标准错误] 命令: %s, 信息: %s',
+            cmd,
+            message,
+          );
+        },
       });
-      cp.on('error', (err) => {
+      const result = await completed;
+      if (result.error) {
         Logger.error(
-          '[schedule][创建任务失败] 命令: %s, 错误信息: %j',
+          '[schedule][执行任务失败] 命令: %s, 错误: %s',
           cmd,
-          err,
+          result.error.message,
         );
-      });
-
-      cp.on('exit', async (code) => {
-        taskLimit.removeQueuedCron(cron.id);
-        Logger.info(
-          '[schedule][执行任务结束] 参数: %s, 退出码: %j',
-          JSON.stringify({
-            ...cron,
-            command: cmd,
-          }),
-          code,
-        );
-        resolve({ ...cron, command: cmd, pid: cp.pid, code });
-      });
-    });
+      }
+      Logger.info(
+        '[schedule][执行任务结束] 任务ID: %s, 退出码: %j',
+        cron.id,
+        result.code,
+      );
+      return { ...cron, command: cmd, pid: cp.pid, ...result } as any;
+    } catch (error) {
+      Logger.error(
+        '[schedule][创建任务失败] 命令: %s, 错误: %s',
+        cmd,
+        asError(error).message,
+      );
+    } finally {
+      taskLimit.removeQueuedCron(cron.id);
+    }
   });
 }

back/shared/schedulerMutationLock.ts

@@ -0,0 +1,38 @@
+import lockfile from 'proper-lockfile';
+import config from '../config';
+
+// HTTP and gRPC both mutate cron definitions. Hold one shared lock from the
+// initial DB read/write through scheduler registration (including rollback).
+// Recovery takes the same lock before reading its replacement snapshot.
+export async function withSchedulerMutation<T>(
+  operation: () => Promise<T>,
+): Promise<T> {
+  let release: () => Promise<void>;
+  try {
+    release = await lockfile.lock(config.crontabFile, {
+      realpath: false,
+      lockfilePath: `${config.crontabFile}.scheduler.lock`,
+      stale: 30000,
+      update: 10000,
+      retries: { retries: 50, factor: 1, minTimeout: 100, maxTimeout: 100 },
+    });
+  } catch (cause) {
+    throw Object.assign(
+      new Error('Scheduler configuration is busy', { cause }),
+      {
+        status: 503,
+      },
+    );
+  }
+  try {
+    return await operation();
+  } finally {
+    await release();
+  }
+}
+
+export function schedulerRegistrationError(message: string, cause: any): Error {
+  return Object.assign(new Error(message, { cause }), {
+    status: cause?.status === 503 ? 503 : 500,
+  });
+}

back/shared/schedulerReadiness.ts

@@ -0,0 +1,77 @@
+// Recovery is single-flight and retried only while unavailable, never at idle.
+export class SchedulerReadiness {
+  private ready = false;
+  private generation = 0;
+  private restore?: () => Promise<void>;
+  private pending?: Promise<boolean>;
+  private retry?: NodeJS.Timeout;
+
+  constructor(private probe: () => Promise<void>, private retryMs = 1000) {}
+
+  configure(restore: () => Promise<void>) {
+    this.restore = restore;
+  }
+
+  invalidate() {
+    this.ready = false;
+    this.generation++;
+    void this.recover();
+  }
+
+  recover(): Promise<boolean> {
+    if (this.pending) return this.pending;
+    if (!this.restore) return Promise.resolve(false);
+    clearTimeout(this.retry);
+    const generation = this.generation;
+    this.ready = false;
+    this.pending = (async () => {
+      try {
+        await this.probe();
+        await this.restore!();
+        await this.probe();
+        this.ready = generation === this.generation;
+      } catch {
+        this.ready = false;
+      }
+      return this.ready;
+    })().finally(() => {
+      this.pending = undefined;
+      if (!this.ready) {
+        this.retry = setTimeout(() => void this.recover(), this.retryMs);
+        this.retry.unref();
+      }
+    });
+    return this.pending;
+  }
+
+  async check(): Promise<boolean> {
+    if (!this.ready) return false;
+    const generation = this.generation;
+    try {
+      await this.probe();
+      return this.ready && generation === this.generation;
+    } catch {
+      this.invalidate();
+      return false;
+    }
+  }
+
+  async ensureReady(timeoutMs = 2000): Promise<void> {
+    let timer: NodeJS.Timeout | undefined;
+    try {
+      const available = await Promise.race([
+        (async () => (await this.check()) || (await this.recover()))(),
+        new Promise<boolean>((resolve) => {
+          timer = setTimeout(() => resolve(false), timeoutMs);
+        }),
+      ]);
+      if (!available) {
+        throw Object.assign(new Error('Scheduler is recovering; try again later'), {
+          status: 503,
+        });
+      }
+    } finally {
+      clearTimeout(timer);
+    }
+  }
+}

back/shared/schemaMigrations.ts

@@ -0,0 +1,70 @@
+import { QueryTypes, Sequelize } from 'sequelize';
+
+// Append new entries; IDs are persisted and must not be renumbered or reused.
+const columns = [
+  {
+    table: 'CrontabViews',
+    column: 'filterRelation',
+    type: 'VARCHAR(255)',
+  },
+  { table: 'Subscriptions', column: 'proxy', type: 'VARCHAR(255)' },
+  { table: 'CrontabViews', column: 'type', type: 'NUMBER' },
+  { table: 'Subscriptions', column: 'autoAddCron', type: 'NUMBER' },
+  { table: 'Subscriptions', column: 'autoDelCron', type: 'NUMBER' },
+  { table: 'Crontabs', column: 'sub_id', type: 'NUMBER' },
+  { table: 'Crontabs', column: 'extra_schedules', type: 'JSON' },
+  { table: 'Crontabs', column: 'task_before', type: 'TEXT' },
+  { table: 'Crontabs', column: 'task_after', type: 'TEXT' },
+  { table: 'Crontabs', column: 'log_name', type: 'VARCHAR(255)' },
+  {
+    table: 'Crontabs',
+    column: 'allow_multiple_instances',
+    type: 'NUMBER',
+  },
+  { table: 'Crontabs', column: 'work_dir', type: 'VARCHAR(255)' },
+  { table: 'Envs', column: 'isPinned', type: 'NUMBER' },
+  { table: 'Envs', column: 'labels', type: 'JSON' },
+  { table: 'Crontabs', column: 'queued_token', type: 'VARCHAR(255)' },
+];
+
+export async function migrateSchema(database: Sequelize): Promise<void> {
+  await database.transaction(async (transaction) => {
+    await database.query(
+      'CREATE TABLE IF NOT EXISTS "SchemaMigrations" ("id" TEXT PRIMARY KEY, "applied_at" TEXT NOT NULL)',
+      { transaction },
+    );
+    const applied = await database.query<{ id: string }>(
+      'SELECT "id" FROM "SchemaMigrations"',
+      { type: QueryTypes.SELECT, transaction },
+    );
+    const appliedIds = new Set(applied.map(({ id }) => id));
+    for (const { table, column, type } of columns) {
+      const id = `add-${table}-${column}`;
+      const fields = await database.query<{ name: string }>(
+        `PRAGMA table_info("${table}")`,
+        {
+          type: QueryTypes.SELECT,
+          transaction,
+        },
+      );
+      if (fields.length === 0)
+        throw new Error(`Migration table is missing: ${table}`);
+      if (!fields.some((field) => field.name === column)) {
+        // table/column/type come only from the static migration manifest above.
+        await database.query(
+          `ALTER TABLE "${table}" ADD COLUMN "${column}" ${type}`,
+          { transaction },
+        );
+      }
+      if (!appliedIds.has(id)) {
+        await database.query(
+          'INSERT INTO "SchemaMigrations" ("id", "applied_at") VALUES (:id, :appliedAt)',
+          {
+            replacements: { id, appliedAt: new Date().toISOString() },
+            transaction,
+          },
+        );
+      }
+    }
+  });
+}