fix: resolve dev panel and CLI regressions

Synchronize scheduler concurrency across API and gRPC processes and restore persisted limits during recovery.

Serialize subscription mutations and preserve shared storage while cleaning Shell and CLI namespaces, historical references, and filesystem-specific naming candidates.

Handle CLI text responses correctly and include script history in backups. Add regression coverage for concurrency, cleanup ownership, long URLs, Unicode paths, and history exports.

back/loaders/initData.ts

@@ -237,7 +237,10 @@ export default async () => {
   } catch { }
 
   // 初始化保存一次ck和定时任务数据
-  cronClient.readiness.configure(() => cronService.autosave_crontab(true));
+  cronClient.readiness.configure(async () => {
+    await cronService.autosave_crontab(true);
+    await systemService.restoreCronConcurrency();
+  });
   await cronClient.readiness.recover();
 
   await envService.set_envs();

back/protos/cron.proto

@@ -5,6 +5,7 @@ package com.ql.cron;
 service Cron {
   rpc addCron(AddCronRequest) returns (AddCronResponse);
   rpc delCron(DeleteCronRequest) returns (DeleteCronResponse);
+  rpc setConcurrency(SetConcurrencyRequest) returns (AddCronResponse);
 }
 
 message ISchedule { string schedule = 1; }
@@ -27,3 +28,6 @@ message AddCronResponse {}
 message DeleteCronRequest { repeated string ids = 1; }
 
 message DeleteCronResponse {}
+
+// Zero restores the default limit.
+message SetConcurrencyRequest { int32 concurrency = 1; }

back/protos/cron.ts

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-ts_proto. DO NOT EDIT.
 // versions:
 //   protoc-gen-ts_proto  v2.6.1
-//   protoc               v3.17.3
+//   protoc               v5.28.3
 // source: back/protos/cron.proto
 
 /* eslint-disable */
@@ -48,6 +48,11 @@ export interface DeleteCronRequest {
 export interface DeleteCronResponse {
 }
 
+/** Zero restores the default limit. */
+export interface SetConcurrencyRequest {
+  concurrency: number;
+}
+
 function createBaseISchedule(): ISchedule {
   return { schedule: "" };
 }
@@ -452,6 +457,64 @@ export const DeleteCronResponse: MessageFns<DeleteCronResponse> = {
   },
 };
 
+function createBaseSetConcurrencyRequest(): SetConcurrencyRequest {
+  return { concurrency: 0 };
+}
+
+export const SetConcurrencyRequest: MessageFns<SetConcurrencyRequest> = {
+  encode(message: SetConcurrencyRequest, writer: BinaryWriter = new BinaryWriter()): BinaryWriter {
+    if (message.concurrency !== 0) {
+      writer.uint32(8).int32(message.concurrency);
+    }
+    return writer;
+  },
+
+  decode(input: BinaryReader | Uint8Array, length?: number): SetConcurrencyRequest {
+    const reader = input instanceof BinaryReader ? input : new BinaryReader(input);
+    let end = length === undefined ? reader.len : reader.pos + length;
+    const message = createBaseSetConcurrencyRequest();
+    while (reader.pos < end) {
+      const tag = reader.uint32();
+      switch (tag >>> 3) {
+        case 1: {
+          if (tag !== 8) {
+            break;
+          }
+
+          message.concurrency = reader.int32();
+          continue;
+        }
+      }
+      if ((tag & 7) === 4 || tag === 0) {
+        break;
+      }
+      reader.skip(tag & 7);
+    }
+    return message;
+  },
+
+  fromJSON(object: any): SetConcurrencyRequest {
+    return { concurrency: isSet(object.concurrency) ? globalThis.Number(object.concurrency) : 0 };
+  },
+
+  toJSON(message: SetConcurrencyRequest): unknown {
+    const obj: any = {};
+    if (message.concurrency !== 0) {
+      obj.concurrency = Math.round(message.concurrency);
+    }
+    return obj;
+  },
+
+  create<I extends Exact<DeepPartial<SetConcurrencyRequest>, I>>(base?: I): SetConcurrencyRequest {
+    return SetConcurrencyRequest.fromPartial(base ?? ({} as any));
+  },
+  fromPartial<I extends Exact<DeepPartial<SetConcurrencyRequest>, I>>(object: I): SetConcurrencyRequest {
+    const message = createBaseSetConcurrencyRequest();
+    message.concurrency = object.concurrency ?? 0;
+    return message;
+  },
+};
+
 export type CronService = typeof CronService;
 export const CronService = {
   addCron: {
@@ -472,11 +535,21 @@ export const CronService = {
     responseSerialize: (value: DeleteCronResponse) => Buffer.from(DeleteCronResponse.encode(value).finish()),
     responseDeserialize: (value: Buffer) => DeleteCronResponse.decode(value),
   },
+  setConcurrency: {
+    path: "/com.ql.cron.Cron/setConcurrency",
+    requestStream: false,
+    responseStream: false,
+    requestSerialize: (value: SetConcurrencyRequest) => Buffer.from(SetConcurrencyRequest.encode(value).finish()),
+    requestDeserialize: (value: Buffer) => SetConcurrencyRequest.decode(value),
+    responseSerialize: (value: AddCronResponse) => Buffer.from(AddCronResponse.encode(value).finish()),
+    responseDeserialize: (value: Buffer) => AddCronResponse.decode(value),
+  },
 } as const;
 
 export interface CronServer extends UntypedServiceImplementation {
   addCron: handleUnaryCall<AddCronRequest, AddCronResponse>;
   delCron: handleUnaryCall<DeleteCronRequest, DeleteCronResponse>;
+  setConcurrency: handleUnaryCall<SetConcurrencyRequest, AddCronResponse>;
 }
 
 export interface CronClient extends Client {
@@ -510,6 +583,21 @@ export interface CronClient extends Client {
     options: Partial<CallOptions>,
     callback: (error: ServiceError | null, response: DeleteCronResponse) => void,
   ): ClientUnaryCall;
+  setConcurrency(
+    request: SetConcurrencyRequest,
+    callback: (error: ServiceError | null, response: AddCronResponse) => void,
+  ): ClientUnaryCall;
+  setConcurrency(
+    request: SetConcurrencyRequest,
+    metadata: Metadata,
+    callback: (error: ServiceError | null, response: AddCronResponse) => void,
+  ): ClientUnaryCall;
+  setConcurrency(
+    request: SetConcurrencyRequest,
+    metadata: Metadata,
+    options: Partial<CallOptions>,
+    callback: (error: ServiceError | null, response: AddCronResponse) => void,
+  ): ClientUnaryCall;
 }
 
 export const CronClient = makeGenericClientConstructor(CronService, "com.ql.cron.Cron") as unknown as {

back/schedule/client.ts

@@ -62,6 +62,25 @@ class Client {
     return this._client;
   }
 
+  async setConcurrency(concurrency: number): Promise<void> {
+    await this.waitForReady(2000);
+    return new Promise((resolve, reject) => {
+      this.client.setConcurrency(
+        { concurrency }, new Metadata(), { deadline: Date.now() + 5000 },
+        (err) => {
+          if (err) {
+            if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) {
+              this.readiness.invalidate();
+              Object.assign(err, { status: 503 });
+            }
+            return reject(err);
+          }
+          resolve();
+        },
+      );
+    });
+  }
+
   async addCron(
     request: AddCronRequest['crons'],
     replace = false

back/schedule/setConcurrency.ts

@@ -0,0 +1,26 @@
+import { ServerUnaryCall, sendUnaryData, status } from '@grpc/grpc-js';
+import os from 'os';
+import { AddCronResponse, SetConcurrencyRequest } from '../protos/cron';
+import taskLimit from '../shared/pLimit';
+
+export async function setConcurrency(
+  call: ServerUnaryCall<SetConcurrencyRequest, AddCronResponse>,
+  callback: sendUnaryData<AddCronResponse>,
+) {
+  const limit = call.request.concurrency;
+  if (!Number.isInteger(limit) || limit < 0 || limit > 2147483647) {
+    callback(
+      Object.assign(new Error('Invalid concurrency'), {
+        code: status.INVALID_ARGUMENT,
+      }),
+      null,
+    );
+    return;
+  }
+  try {
+    await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
+    callback(null, {});
+  } catch (error) {
+    callback(error as Error, null);
+  }
+}

back/services/grpc.ts

@@ -4,6 +4,7 @@ import { HealthService } from '../protos/health';
 import { ApiService } from '../protos/api';
 import { addCron } from '../schedule/addCron';
 import { delCron } from '../schedule/delCron';
+import { setConcurrency } from '../schedule/setConcurrency';
 import { check } from '../schedule/health';
 import * as Api from '../schedule/api';
 import Logger from '../loaders/logger';
@@ -27,7 +28,7 @@ export class GrpcServerService {
   async initialize() {
     try {
       this.server.addService(HealthService, { check });
-      this.server.addService(CronService, { addCron, delCron });
+      this.server.addService(CronService, { addCron, delCron, setConcurrency });
       this.server.addService(ApiService, Api);
 
       const tlsConfig = await initGrpcCerts();

back/services/subscription.ts

@@ -30,11 +30,14 @@ import dayjs from 'dayjs';
 import { LOG_END_SYMBOL } from '../config/const';
 import { formatCommand, formatUrl } from '../config/subscription';
 import { CrontabModel } from '../data/cron';
-import {
-  assertSubscriptionAlias,
-  resolveSubscriptionPath,
-} from '../shared/subscriptionPath';
+import { assertSubscriptionAlias } from '../shared/subscriptionPath';
 import { resolveLogPath } from '../shared/logPath';
+import {
+  getSubscriptionStorageCandidates,
+  isSubscriptionStorageName,
+  resolveSubscriptionStoragePath,
+} from '../shared/subscriptionStorage';
+import { withSubscriptionMutation } from '../shared/subscriptionMutationLock';
 import CrontabService from './cron';
 import taskLimit from '../shared/pLimit';
 import { logStreamManager } from '../shared/logStreamManager';
@@ -230,11 +233,13 @@ export default class SubscriptionService {
 
   public async create(payload: Subscription): Promise<Subscription> {
     assertSubscriptionAlias(payload.alias);
-    const tab = new Subscription(payload);
-    const doc = await this.insert(tab);
-    await this.handleTask(doc.get({ plain: true }));
-    await this.setSshConfig();
-    return doc;
+    return withSubscriptionMutation(async () => {
+      const tab = new Subscription(payload);
+      const doc = await this.insert(tab);
+      await this.handleTask(doc.get({ plain: true }));
+      await this.setSshConfig();
+      return doc;
+    });
   }
 
   public async insert(payload: Subscription): Promise<SubscriptionInstance> {
@@ -243,13 +248,15 @@ export default class SubscriptionService {
 
   public async update(payload: Subscription): Promise<Subscription> {
     assertSubscriptionAlias(payload.alias);
-    const doc = await this.getDb({ id: payload.id });
-    const tab = new Subscription({ ...doc, ...payload });
-    const newDoc = await this.updateDb(tab);
-    await this.handleTask(newDoc, !newDoc.is_disabled);
-    await this.removeSshConfigForSubscription(doc);
-    await this.setSshConfig();
-    return newDoc;
+    return withSubscriptionMutation(async () => {
+      const doc = await this.getDb({ id: payload.id });
+      const tab = new Subscription({ ...doc, ...payload });
+      const newDoc = await this.updateDb(tab);
+      await this.handleTask(newDoc, !newDoc.is_disabled);
+      await this.removeSshConfigForSubscription(doc);
+      await this.setSshConfig();
+      return newDoc;
+    });
   }
 
   private async removeSshConfigForSubscription(doc: Subscription) {
@@ -301,35 +308,71 @@ export default class SubscriptionService {
   }
 
   public async remove(ids: number[], query: { force?: boolean }) {
-    const docs = await SubscriptionModel.findAll({ where: { id: ids } });
-    // Validate all deletion targets before changing rows or removing any files.
-    if (query?.force === true) {
-      for (const doc of docs) {
-        resolveSubscriptionPath(config.scriptPath, doc.alias);
-        resolveSubscriptionPath(config.repoPath, doc.alias);
+    return withSubscriptionMutation(async () => {
+      const docs = await SubscriptionModel.findAll({ where: { id: ids } });
+      const storagePaths: string[] = [];
+      const retainedPaths: string[] = [];
+      const collectPaths = (
+        doc: Subscription,
+        targets: string[],
+        deleting = false,
+      ) => {
+        for (const names of getSubscriptionStorageCandidates(doc)) {
+          // A URL accepted by one engine can produce an invalid name in another.
+          // Validate each candidate independently, retaining the safe candidate.
+          if (!Object.values(names).every(isSubscriptionStorageName)) continue;
+          const resolve = deleting ? resolveSubscriptionStoragePath : path.resolve;
+          const scriptPath = resolve(config.scriptPath, names.script);
+          if (scriptPath) targets.push(scriptPath);
+          if (names.repo) {
+            const repoPath = resolve(config.repoPath, names.repo);
+            if (repoPath) targets.push(repoPath);
+          }
+          if (names.raw) {
+            const rawPath = resolve(path.join(config.dataPath, 'raw'), names.raw);
+            if (rawPath) targets.push(rawPath);
+          }
+        }
+      };
+      // Validate all deletion targets before changing rows or removing any files.
+      if (query?.force === true) {
+        for (const doc of docs) {
+          assertSubscriptionAlias(doc.alias);
+          resolveSubscriptionStoragePath(config.scriptPath, doc.alias);
+          resolveSubscriptionStoragePath(config.repoPath, doc.alias);
+          collectPaths(doc, storagePaths, true);
+        }
+        const remaining = await SubscriptionModel.findAll({ where: {} });
+        for (const doc of remaining) {
+          if (!ids.includes(doc.id!)) collectPaths(doc, retainedPaths);
+        }
       }
-    }
-    for (const doc of docs) {
-      await this.handleTask(doc.get({ plain: true }), false);
-    }
-    await SubscriptionModel.destroy({ where: { id: ids } });
-    for (const doc of docs) {
-      await this.removeSshConfigForSubscription(doc);
-    }
-    await this.setSshConfig();
+      for (const doc of docs) {
+        await this.handleTask(doc.get({ plain: true }), false);
+      }
+      await SubscriptionModel.destroy({ where: { id: ids } });
+      for (const doc of docs) {
+        await this.removeSshConfigForSubscription(doc);
+      }
+      await this.setSshConfig();
 
-    if (query?.force === true) {
-      const crons = await CrontabModel.findAll({ where: { sub_id: ids } });
-      if (crons?.length) {
-        await this.crontabService.remove(crons.map((x) => x.id!));
+      if (query?.force === true) {
+        const crons = await CrontabModel.findAll({ where: { sub_id: ids } });
+        if (crons?.length) {
+          await this.crontabService.remove(crons.map((x) => x.id!));
+        }
+        for (const storagePath of new Set(storagePaths)) {
+          // rm is recursive: an ancestor must also be retained if a surviving
+          // subscription uses a nested branch, and vice versa.
+          const shared = retainedPaths.some((retained) =>
+            retained === storagePath ||
+            retained.startsWith(`${storagePath}${path.sep}`) ||
+            storagePath.startsWith(`${retained}${path.sep}`),
+          );
+          if (!shared) await rmPath(storagePath);
+        }
       }
-      for (const doc of docs) {
-        const filePath = resolveSubscriptionPath(config.scriptPath, doc.alias);
-        const repoPath = resolveSubscriptionPath(config.repoPath, doc.alias);
-        await rmPath(filePath);
-        await rmPath(repoPath);
-      }
-    }
+    });
   }
 
   public async getDb(

back/services/system.ts

@@ -34,6 +34,8 @@ import {
   SystemModelInfo,
 } from '../data/system';
 import taskLimit from '../shared/pLimit';
+import cronClient from '../schedule/client';
+import { withSchedulerMutation } from '../shared/schedulerMutationLock';
 import NotificationService from './notify';
 import ScheduleService, { TaskCallbacks } from './schedule';
 import SockService from './sock';
@@ -122,15 +124,29 @@ export default class SystemService {
   }
 
   public async updateCronConcurrency(info: SystemModelInfo) {
-    const oDoc = await this.getSystemConfig();
-    await this.updateAuthDb({
-      ...oDoc,
-      info: { ...oDoc.info, ...info },
-    });
-    if (info.cronConcurrency) {
-      await taskLimit.setCustomLimit(info.cronConcurrency);
+    const limit = info.cronConcurrency ?? 0;
+    if (!Number.isInteger(limit) || limit < 0 || limit > 2147483647) {
+      throw Object.assign(new Error(t('参数错误')), { status: 400 });
     }
-    return { code: 200, data: info };
+    return withSchedulerMutation(async () => {
+      const oDoc = await this.getSystemConfig();
+      await this.updateAuthDb({
+        ...oDoc,
+        info: { ...oDoc.info, ...info, cronConcurrency: limit },
+      });
+      await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
+      await cronClient.setConcurrency(limit);
+      return { code: 200, data: info };
+    });
+  }
+
+  public async restoreCronConcurrency() {
+    await withSchedulerMutation(async () => {
+      const doc = await this.getSystemConfig();
+      const limit = doc.info?.cronConcurrency ?? 0;
+      await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
+      await cronClient.setConcurrency(limit);
+    });
   }
 
   public async updateDependenceProxy(info: SystemModelInfo) {
@@ -445,10 +461,15 @@ export default class SystemService {
         dataDirs = dataDirs.concat(type.filter((x) => x !== 'base'));
       }
       const allowed = new Set(['db', 'upload', 'config', 'scripts', 'log', 'deps',
-        'syslog', 'dep_cache', 'raw', 'repo', 'ssh.d']);
+        'syslog', 'dep_cache', 'raw', 'repo', 'ssh.d', 'script-history']);
       if (dataDirs.some((dir) => !allowed.has(dir))) {
         return res.status(400).send({ code: 400, message: t('参数错误') });
       }
+      if (dataDirs.includes('scripts')) dataDirs.push('script-history');
+      // History is created lazily; old/fresh installations must still export.
+      if (!fs.existsSync(path.join(config.dataPath, 'script-history'))) {
+        dataDirs = dataDirs.filter((dir) => dir !== 'script-history');
+      }
       const dataPaths = [...new Set(dataDirs)].map((dir) => `data/${dir}`);
       await promisify(execFile)('tar', ['-zcf', config.dataTgzFile, '--', ...dataPaths], {
         cwd: path.dirname(config.dataPath),

back/shared/subscriptionMutationLock.ts

@@ -0,0 +1,30 @@
+import lockfile from 'proper-lockfile';
+import path from 'path';
+import config from '../config';
+
+// Protect subscription storage ownership from the first database read through
+// cleanup. Creation and URL/branch updates must use the same cross-process lock.
+export async function withSubscriptionMutation<T>(
+  operation: () => Promise<T>,
+): Promise<T> {
+  const target = `${path.resolve(config.scriptPath)}.subscriptions`;
+  let release: () => Promise<void>;
+  try {
+    release = await lockfile.lock(target, {
+      realpath: false,
+      stale: 30000,
+      update: 10000,
+      retries: { retries: 50, factor: 1, minTimeout: 100, maxTimeout: 100 },
+    });
+  } catch (cause) {
+    throw Object.assign(
+      new Error('Subscription configuration is busy', { cause }),
+      { status: 503 },
+    );
+  }
+  try {
+    return await operation();
+  } finally {
+    await release();
+  }
+}

back/shared/subscriptionStorage.ts

@@ -0,0 +1,106 @@
+import type { Subscription } from '../data/subscription';
+import { formatUrl } from '../config/subscription';
+import path from 'path';
+import fs from 'fs';
+import { resolveFileAccess } from './fileAccess';
+
+// Storage namespaces come from URLs and branches, not user aliases. In
+// particular, a legal Git directory can exceed the alias's 200-character limit.
+export function isSubscriptionStorageName(name: string): boolean {
+  return (
+    typeof name === 'string' &&
+    !!name &&
+    !path.isAbsolute(name) &&
+    !/[\0\r\n\\]/.test(name) &&
+    !name.split('/').some((part) => !part || part === '.' || part === '..')
+  );
+}
+
+export function resolveSubscriptionStoragePath(
+  root: string,
+  name: string,
+): string | undefined {
+  const target = isSubscriptionStorageName(name)
+    ? resolveFileAccess(root, [name])
+    : '';
+  if (!target) {
+    if (isSubscriptionStorageName(name)) {
+      try {
+        fs.lstatSync(path.resolve(root, name));
+      } catch (error: any) {
+        // A long query can make the Shell candidate unrepresentable while
+        // the CLI's pathname-based file exists. Use the actual filesystem
+        // limit: macOS permits Unicode names exceeding 255 UTF-8 bytes.
+        if (error.code === 'ENAMETOOLONG') return undefined;
+      }
+    }
+    throw Object.assign(new Error('Invalid subscription path'), {
+      status: 400,
+    });
+  }
+  return target;
+}
+
+/** Match shell/update.sh get_uniq_path and update_raw, including SCP URLs. */
+export function getSubscriptionStorageNames(
+  doc: Subscription,
+  engine: 'shell' | 'cli' = 'shell',
+): {
+  script: string;
+  repo?: string;
+  raw?: string;
+} {
+  // Old imported rows may only have an alias and no URL.
+  if (!doc.url) return { script: doc.alias, repo: doc.alias };
+  const url = formatUrl(doc).url!;
+  const trimmed = url.endsWith('/') ? url.slice(0, -1) : url;
+  const basename = trimmed.slice(trimmed.lastIndexOf('/') + 1);
+  const dot = basename.lastIndexOf('.');
+  const repo = dot < 0 ? basename : basename.slice(0, dot);
+  const parentUrl = engine === 'cli' ? trimmed : url;
+  const slash = parentUrl.lastIndexOf('/');
+  const parent =
+    engine === 'shell' && slash < 0 ? url : parentUrl.slice(0, slash);
+  const owner = parent.slice(parent.lastIndexOf('/') + 1);
+  const author = owner.slice(owner.lastIndexOf(':') + 1);
+  const stem = `${author.slice(author.lastIndexOf('.') + 1)}_${repo}`;
+  if (doc.type === 'file') {
+    const suffix =
+      engine === 'cli'
+        ? path.extname(new URL(url).pathname)
+        : `.${url.slice(url.lastIndexOf('.') + 1)}`;
+    const raw = `${stem}${suffix}`;
+    return { script: `raw_${raw}`, raw };
+  }
+  const directory = doc.branch ? `${stem}_${doc.branch}` : stem;
+  return { script: directory, repo: directory };
+}
+
+// A subscription may have run through either engine before being deleted.
+export function getSubscriptionStorageCandidates(doc: Subscription) {
+  const candidates: ReturnType<typeof getSubscriptionStorageNames>[] = [];
+  for (const engine of ['shell', 'cli'] as const) {
+    try {
+      candidates.push(getSubscriptionStorageNames(doc, engine));
+    } catch {
+      // Historical rows may have an invalid SSH alias or missing credentials.
+      // Retain their URL-derived storage without needing an executable SSH URL.
+      try {
+        candidates.push(
+          getSubscriptionStorageNames(
+            {
+              url: doc.url,
+              alias: doc.alias,
+              branch: doc.branch,
+              type: doc.type === 'private-repo' ? 'public-repo' : doc.type,
+            },
+            engine,
+          ),
+        );
+      } catch {
+        // One engine's invalid URL must not hide the other engine's candidate.
+      }
+    }
+  }
+  return candidates;
+}