From 24f7d1a46f8754f8570cf0c682a192fe9d190985 Mon Sep 17 00:00:00 2001 From: whyour Date: Wed, 7 Oct 2026 12:35:10 +0800 Subject: [PATCH] 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 | 5 +- back/protos/cron.proto | 4 + back/protos/cron.ts | 90 ++- back/schedule/client.ts | 19 + back/schedule/setConcurrency.ts | 26 + back/services/grpc.ts | 3 +- back/services/subscription.ts | 127 ++-- back/services/system.ts | 39 +- back/shared/subscriptionMutationLock.ts | 30 + back/shared/subscriptionStorage.ts | 106 ++++ cli/src/remote/commands/open.ts | 2 +- cli/test/remote/openapi.test.cjs | 26 + test/back/advisory-input-security.test.cjs | 1 + test/back/grpc-mtls-security.test.cjs | 1 + test/back/listen-security.test.cjs | 1 + test/back/scheduler-concurrency.test.cjs | 411 +++++++++++++ test/back/subscription-mutation-lock.test.cjs | 83 +++ .../subscription-review-regression.test.cjs | 4 +- .../subscription-storage-regression.test.cjs | 553 ++++++++++++++++++ test/back/system-dependence-cache.test.cjs | 1 + test/back/system-export-history.test.cjs | 119 ++++ test/back/wpush-grpc.test.cjs | 1 + 22 files changed, 1595 insertions(+), 57 deletions(-) create mode 100644 back/schedule/setConcurrency.ts create mode 100644 back/shared/subscriptionMutationLock.ts create mode 100644 back/shared/subscriptionStorage.ts create mode 100644 test/back/scheduler-concurrency.test.cjs create mode 100644 test/back/subscription-mutation-lock.test.cjs create mode 100644 test/back/subscription-storage-regression.test.cjs create mode 100644 test/back/system-export-history.test.cjs diff --git a/back/loaders/initData.ts b/back/loaders/initData.ts index 8854841a..2620f58e 100644 --- a/back/loaders/initData.ts +++ b/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(); diff --git a/back/protos/cron.proto b/back/protos/cron.proto index 505eabe6..539f6435 100644 --- a/back/protos/cron.proto +++ b/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; } diff --git a/back/protos/cron.ts b/back/protos/cron.ts index 75832810..3618725f 100644 --- a/back/protos/cron.ts +++ b/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 = { }, }; +function createBaseSetConcurrencyRequest(): SetConcurrencyRequest { + return { concurrency: 0 }; +} + +export const SetConcurrencyRequest: MessageFns = { + 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>>(base?: I): SetConcurrencyRequest { + return SetConcurrencyRequest.fromPartial(base ?? ({} as any)); + }, + fromPartial, 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; delCron: handleUnaryCall; + setConcurrency: handleUnaryCall; } export interface CronClient extends Client { @@ -510,6 +583,21 @@ export interface CronClient extends Client { options: Partial, 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, + callback: (error: ServiceError | null, response: AddCronResponse) => void, + ): ClientUnaryCall; } export const CronClient = makeGenericClientConstructor(CronService, "com.ql.cron.Cron") as unknown as { diff --git a/back/schedule/client.ts b/back/schedule/client.ts index 3243e806..9089c78e 100644 --- a/back/schedule/client.ts +++ b/back/schedule/client.ts @@ -62,6 +62,25 @@ class Client { return this._client; } + async setConcurrency(concurrency: number): Promise { + 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 diff --git a/back/schedule/setConcurrency.ts b/back/schedule/setConcurrency.ts new file mode 100644 index 00000000..20fb1876 --- /dev/null +++ b/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, + callback: sendUnaryData, +) { + 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); + } +} diff --git a/back/services/grpc.ts b/back/services/grpc.ts index 2e8c6bb6..c1c46c2b 100644 --- a/back/services/grpc.ts +++ b/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(); diff --git a/back/services/subscription.ts b/back/services/subscription.ts index a2e153ba..225b5b24 100644 --- a/back/services/subscription.ts +++ b/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 { 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 { @@ -243,13 +248,15 @@ export default class SubscriptionService { public async update(payload: Subscription): Promise { 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( diff --git a/back/services/system.ts b/back/services/system.ts index a8538b16..b5725d78 100644 --- a/back/services/system.ts +++ b/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), diff --git a/back/shared/subscriptionMutationLock.ts b/back/shared/subscriptionMutationLock.ts new file mode 100644 index 00000000..09d1a272 --- /dev/null +++ b/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( + operation: () => Promise, +): Promise { + const target = `${path.resolve(config.scriptPath)}.subscriptions`; + let release: () => Promise; + 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(); + } +} diff --git a/back/shared/subscriptionStorage.ts b/back/shared/subscriptionStorage.ts new file mode 100644 index 00000000..156e49bb --- /dev/null +++ b/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[] = []; + 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; +} diff --git a/cli/src/remote/commands/open.ts b/cli/src/remote/commands/open.ts index dd20c794..f39209b3 100644 --- a/cli/src/remote/commands/open.ts +++ b/cli/src/remote/commands/open.ts @@ -241,7 +241,7 @@ export async function openCommand( body, authenticated: !op.anonymous, output: values.output as string | undefined, - text: op.path === 'system/log', + text: ['system/log', 'system/config/node-mirror'].includes(op.path), timeoutMs: values.timeout ? Number(values.timeout) * 1000 : undefined, }); const data = diff --git a/cli/test/remote/openapi.test.cjs b/cli/test/remote/openapi.test.cjs index 8c872cbc..66513cb0 100644 --- a/cli/test/remote/openapi.test.cjs +++ b/cli/test/remote/openapi.test.cjs @@ -211,3 +211,29 @@ test('owner two-factor challenge remains actionable without exposing server mess assert.equal(result.code, 3); assert.match(result.err, /two-factor-login/); assert.doesNotMatch(result.err, /SERVER-SECRET/); assert.equal(f.requests.length, 1); assert.equal(f.requests[0].headers.authorization, undefined); }); + +test('node mirror configuration accepts the real empty response for named and raw API commands', async t => { + const f = await fixture(t); + f.reply((_req, res) => { res.writeHead(200, { 'content-type': 'application/octet-stream' }); res.end(); }); + for (const args of [ + ['system', 'config-node-mirror'], + ['api', 'request', 'PUT', '/open/system/config/node-mirror'], + ]) { + const reply = success(await f.run([...args, '--data', '{"nodeMirror":"https://registry.npmmirror.com"}'])); + assert.equal(reply.code, 200); + assert.equal(f.requests.at(-1).method, 'PUT'); + assert.equal(f.requests.at(-1).url, '/panel/open/system/config/node-mirror'); + } + assert.equal(f.requests.length, 2, 'configuration writes must not be retried'); +}); + +test('empty node mirror responses still reject authentication and server failures without retries', async t => { + const f = await fixture(t); + for (const status of [401, 403, 500]) { + f.reply((_req, res) => { res.writeHead(status, { 'content-type': 'application/octet-stream' }); res.end(); }); + const result = await f.run(['system', 'config-node-mirror', '--data', '{}']); + assert.notEqual(result.code, 0); + assert.equal(result.out, ''); + } + assert.equal(f.requests.length, 3); +}); diff --git a/test/back/advisory-input-security.test.cjs b/test/back/advisory-input-security.test.cjs index 79477f1b..d4318b52 100644 --- a/test/back/advisory-input-security.test.cjs +++ b/test/back/advisory-input-security.test.cjs @@ -267,6 +267,7 @@ function systemService(config, overrides = {}) { }, '../data/system': {}, '../shared/pLimit': {}, + '../schedule/client': {}, '../shared/i18n': { ...translate, setLang() {} }, './notify': {}, './schedule': {}, diff --git a/test/back/grpc-mtls-security.test.cjs b/test/back/grpc-mtls-security.test.cjs index d574d761..6c60f2d7 100644 --- a/test/back/grpc-mtls-security.test.cjs +++ b/test/back/grpc-mtls-security.test.cjs @@ -92,6 +92,7 @@ test('real gRPC server rejects plaintext and TLS without client certificates, ac }, '../schedule/addCron': {}, '../schedule/delCron': {}, + '../schedule/setConcurrency': {}, '../schedule/api': {}, '../schedule/health': { check: (_call, cb) => { diff --git a/test/back/listen-security.test.cjs b/test/back/listen-security.test.cjs index f21092ab..2f73cf55 100644 --- a/test/back/listen-security.test.cjs +++ b/test/back/listen-security.test.cjs @@ -48,6 +48,7 @@ for (const host of ['127.0.0.1', '::1', '192.0.2.1', '::']) { '../protos/api': { ApiService: {} }, '../schedule/addCron': {}, '../schedule/delCron': {}, + '../schedule/setConcurrency': {}, '../schedule/health': {}, '../schedule/api': {}, '../config/grpcCerts': { diff --git a/test/back/scheduler-concurrency.test.cjs b/test/back/scheduler-concurrency.test.cjs new file mode 100644 index 00000000..99b4ef9f --- /dev/null +++ b/test/back/scheduler-concurrency.test.cjs @@ -0,0 +1,411 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const grpc = require('@grpc/grpc-js'); +const load = require('../helpers/load-security-module.cjs'); +const logger = { info() {}, debug() {}, warn() {}, error() {} }; +const tick = () => new Promise(setImmediate); +function createLimit() { + return load('back/shared/pLimit.ts', { + '../data/system': { + AuthDataType: { systemConfig: 'systemConfig' }, + SystemModel: { + sync: async () => {}, + findOne: async () => ({ info: { cronConcurrency: 1 } }), + }, + }, + '../loaders/logger': logger, + '../services/notify': class {}, + '../shared/i18n': { t: (x) => x, tf: (x) => x }, + '../config': {}, + '../config/grpcCerts': { getGrpcCerts: () => null }, + '../protos/api': {}, + }).default; +} +function createSystem(client, limit, options = {}) { + const store = options.store || { + state: { + id: 1, + type: 'systemConfig', + info: { cronConcurrency: 1, timezone: 'Asia/Shanghai' }, + }, + }; + const writes = []; + const Service = load('back/services/system.ts', { + '../config': options.config || {}, + '../shared/schedulerMutationLock': options.config + ? load('back/shared/schedulerMutationLock.ts', { + '../config': options.config, + }) + : { withSchedulerMutation: (operation) => operation() }, + '../config/const': {}, + '../config/util': {}, + '../data/dependence': {}, + '../data/notify': {}, + '../data/system': { + AuthDataType: { systemConfig: 'systemConfig' }, + SystemModel: { + findOne: async () => ({ + ...store.state, + get: () => ({ ...store.state }), + }), + update: async (values) => { + writes.push(values); + store.state = { ...store.state, ...values }; + }, + }, + }, + '../shared/pLimit': limit, + '../schedule/client': client, + '../shared/i18n': { t: (x) => x }, + './notify': class {}, + './schedule': class {}, + './sock': class {}, + }).default; + return { + service: new Service(logger, {}, {}), + writes, + state: () => store.state, + }; +} +async function rpcFixture(t) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ql-concurrency-rpc-')); + t.after(() => fs.rmSync(root, { recursive: true, force: true })); + const certs = await load('back/config/grpcCerts.ts', { + './index': { configPath: root }, + './util': { fileExist: async (p) => fs.existsSync(p) }, + '../loaders/logger': logger, + }).initGrpcCerts(); + const schedulerLimit = createLimit(), + apiLimit = createLimit(); + await tick(); + const handler = load('back/schedule/setConcurrency.ts', { + '../shared/pLimit': schedulerLimit, + }); + let port; + const Grpc = load('back/services/grpc.ts', { + '../config': { bindHostGrpc: '127.0.0.1', grpcPort: 0 }, + '../config/grpcCerts': { initGrpcCerts: async () => certs }, + '../loaders/logger': logger, + './metrics': { metricsService: { record() {} } }, + '../schedule/addCron': {}, + '../schedule/delCron': {}, + '../schedule/api': {}, + '../schedule/health': { check: (_call, cb) => cb(null, { status: 1 }) }, + '../schedule/setConcurrency': handler, + '@grpc/grpc-js': { + ...grpc, + Server: class extends grpc.Server { + bindAsync(address, creds, cb) { + super.bindAsync(address, creds, (error, boundPort) => { + port = boundPort; + cb(error, boundPort); + }); + } + }, + }, + }).GrpcServerService; + const server = new Grpc(); + await server.initialize(); + const client = load('back/schedule/client.ts', { + '../config': { grpcPort: port }, + '../config/grpcCerts': { getGrpcCerts: () => certs }, + }).default; + t.after(async () => { + client._client?.close(); + await server.shutdown(); + }); + return { + ...createSystem(client, apiLimit), + schedulerLimit, + apiLimit, + client, + }; +} +function enqueue(t, limit, count) { + const releases = [], + runs = []; + for (let i = 0; i < count; i++) { + let release; + const gate = new Promise((resolve) => { + release = resolve; + }); + releases.push(release); + runs.push(limit.runWithCronLimit({ id: String(i) }, () => gate)); + } + t.after(async () => { + releases.forEach((release) => release()); + await Promise.all(runs); + }); + return { releases, runs }; +} + +test('concurrency changes reach the independent scheduler queue over mTLS immediately and preserve running jobs', async (t) => { + const f = await rpcFixture(t); + const { releases, runs } = enqueue(t, f.schedulerLimit, 3); + await tick(); + assert.equal(f.schedulerLimit.cronLimitActiveCount, 1); + assert.equal(f.schedulerLimit.cronLimitPendingCount, 2); + assert.equal( + (await f.service.updateCronConcurrency({ cronConcurrency: 2 })).code, + 200, + ); + await tick(); + assert.equal(f.schedulerLimit.cronLimitActiveCount, 2); + assert.equal(f.schedulerLimit.cronLimitPendingCount, 1); + await f.service.updateCronConcurrency({ cronConcurrency: 1 }); + assert.equal( + f.schedulerLimit.cronLimitActiveCount, + 2, + 'lowering the limit must not stop active jobs', + ); + releases[0](); + await runs[0]; + await tick(); + assert.equal(f.schedulerLimit.cronLimitActiveCount, 1); + assert.equal(f.schedulerLimit.cronLimitPendingCount, 1); + releases[1](); + await runs[1]; + await tick(); + assert.equal(f.schedulerLimit.cronLimitActiveCount, 1); + assert.equal(f.schedulerLimit.cronLimitPendingCount, 0); + releases[2](); + await runs[2]; + assert.equal(f.state().info.cronConcurrency, 1); +}); + +test('zero, null and omitted concurrency restore the default in both API and scheduler queues without a restart', async (t) => { + const f = await rpcFixture(t); + for (const value of [0, null, undefined]) { + await f.service.updateCronConcurrency({ cronConcurrency: 1 }); + const scheduler = enqueue(t, f.schedulerLimit, 4), + manual = enqueue(t, f.apiLimit, 4); + await tick(); + assert.equal(f.schedulerLimit.cronLimitActiveCount, 1); + assert.equal(f.apiLimit.cronLimitActiveCount, 1); + await f.service.updateCronConcurrency( + value === undefined ? {} : { cronConcurrency: value }, + ); + await tick(); + assert.equal( + f.state().info.cronConcurrency, + 0, + 'persist the same default that both queues apply', + ); + assert.equal(f.schedulerLimit.cronLimitActiveCount, 4); + assert.equal(f.apiLimit.cronLimitActiveCount, 4); + [...scheduler.releases, ...manual.releases].forEach((release) => release()); + await Promise.all([...scheduler.runs, ...manual.runs]); + } +}); + +test('invalid concurrency is rejected before configuration writes or queue changes', async () => { + const mutations = []; + const f = createSystem( + { setConcurrency: async (value) => mutations.push(value) }, + { setCustomLimit: async (value) => mutations.push(value) }, + ); + for (const value of [-1, 1.5, NaN, Infinity, 2147483648]) + await assert.rejects( + f.service.updateCronConcurrency({ cronConcurrency: value }), + { status: 400 }, + ); + assert.deepEqual(f.writes, []); + assert.deepEqual(mutations, []); + const handler = load('back/schedule/setConcurrency.ts', { + '../shared/pLimit': { + setCustomLimit: async (value) => mutations.push(value), + }, + }).setConcurrency; + for (const value of [-1, 0.5, 2147483648]) + await handler({ request: { concurrency: value } }, (error) => + assert.equal(error.code, grpc.status.INVALID_ARGUMENT), + ); + assert.deepEqual(mutations, []); +}); + +for (const code of [grpc.status.UNAVAILABLE, grpc.status.DEADLINE_EXCEEDED]) { + test(`concurrency RPC failure ${code} invalidates readiness and is not reported as success or retried`, async () => { + let writes = 0, + invalidations = 0; + const fake = { + waitForReady: (_deadline, callback) => callback(), + setConcurrency: (_request, _metadata, options, callback) => { + assert.ok(options.deadline > Date.now()); + writes++; + callback(Object.assign(new Error('scheduler unavailable'), { code })); + }, + }; + const client = load('back/schedule/client.ts', { + '../protos/cron': { + CronClient: class { + constructor() { + return fake; + } + }, + }, + '../config': { grpcPort: 1 }, + '../config/grpcCerts': { + getGrpcCerts: () => ({ + caCert: 'ca', + clientKey: 'key', + clientCert: 'cert', + }), + }, + '@grpc/grpc-js': { ...grpc, credentials: { createSsl: () => ({}) } }, + }).default; + client.readiness.invalidate = () => invalidations++; + const f = createSystem(client, { setCustomLimit: async () => {} }); + await assert.rejects( + f.service.updateCronConcurrency({ cronConcurrency: 2 }), + { status: 503 }, + ); + assert.equal(writes, 1); + assert.equal(invalidations, 1); + }); +} + +function mutationConfig(t) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ql-concurrency-lock-')); + t.after(() => fs.rmSync(root, { recursive: true, force: true })); + return { crontabFile: path.join(root, 'crontab.list') }; +} + +test('independent API service instances serialize database and scheduler updates with the shared filesystem lock', async (t) => { + const config = mutationConfig(t); + const store = { state: { id: 1, info: { cronConcurrency: 1 } } }; + let release, entered; + const firstEntered = new Promise((resolve) => (entered = resolve)); + const gate = new Promise((resolve) => (release = resolve)); + t.after(() => release()); + let scheduler = 1, + api = 1; + const client = { + setConcurrency: async (value) => { + if (value === 10) { + entered(); + await gate; + } + scheduler = value; + }, + }; + const limit = { setCustomLimit: async (value) => (api = value) }; + const a = createSystem(client, limit, { config, store }); + const b = createSystem(client, limit, { config, store }); + const first = a.service.updateCronConcurrency({ cronConcurrency: 10 }); + await firstEntered; + const second = b.service.updateCronConcurrency({ cronConcurrency: 1 }); + await new Promise((resolve) => setTimeout(resolve, 150)); + assert.equal( + b.writes.length, + 0, + 'a second writer cannot persist ahead of the first RPC', + ); + release(); + await Promise.all([first, second]); + assert.deepEqual( + [store.state.info.cronConcurrency, api, scheduler], + [1, 1, 1], + ); +}); + +for (const newerLimit of [undefined, 2]) { + test(`production recovery restores persisted concurrency before health becomes ready (newer limit: ${newerLimit})`, async (t) => { + const config = mutationConfig(t); + const store = { state: { id: 1, info: { cronConcurrency: 1 } } }; + let scheduler = 1, + api = 1, + fail = false, + restoreJobs = 0; + let unblockRecovery, recoveryEntered; + const entered = new Promise((resolve) => (recoveryEntered = resolve)); + const gate = new Promise((resolve) => (unblockRecovery = resolve)); + t.after(() => unblockRecovery()); + const fake = { + waitForReady: (_deadline, callback) => callback(), + makeUnaryRequest: (...args) => args.at(-1)(null, { status: 1 }), + setConcurrency: ({ concurrency }, _metadata, _options, callback) => { + if (fail) { + fail = false; + callback( + Object.assign(new Error('lost RPC'), { + code: grpc.status.UNAVAILABLE, + }), + ); + } else { + scheduler = concurrency; + callback(null, {}); + } + }, + }; + const client = load('back/schedule/client.ts', { + '../protos/cron': { + CronClient: class { + constructor() { + return fake; + } + }, + }, + '../config': { grpcPort: 1 }, + '../config/grpcCerts': { + getGrpcCerts: () => ({ + caCert: 'ca', + clientKey: 'key', + clientCert: 'cert', + }), + }, + '@grpc/grpc-js': { ...grpc, credentials: { createSsl: () => ({}) } }, + }).default; + const f = createSystem( + client, + { setCustomLimit: async (value) => (api = value) }, + { config, store }, + ); + const source = fs.readFileSync('back/loaders/initData.ts', 'utf8'); + const callback = source.match( + /cronClient\.readiness\.configure\((async \(\) => \{[\s\S]*?\n \})\);/, + )[1]; + const restore = new Function( + 'cronService', + 'systemService', + `return ${callback}`, + )( + { + autosave_crontab: async (strict) => { + assert.equal(strict, true); + restoreJobs++; + if (restoreJobs > 1) { + recoveryEntered(); + await gate; + } + }, + }, + f.service, + ); + client.readiness.configure(restore); + assert.equal(await client.readiness.recover(), true); + fail = true; + await assert.rejects( + f.service.updateCronConcurrency({ cronConcurrency: 10 }), + { status: 503 }, + ); + await entered; + assert.equal(await client.readiness.check(), false); + assert.equal(scheduler, 1); + // A newer successful request must win over the failed request being recovered. + if (newerLimit !== undefined) + await f.service.updateCronConcurrency({ cronConcurrency: newerLimit }); + scheduler = 1; // Simulate the scheduler restarting while recovery is in progress. + const recovered = client.readiness.recover(); + unblockRecovery(); + assert.equal(await recovered, true); + const expected = newerLimit ?? 10; + assert.deepEqual( + [store.state.info.cronConcurrency, api, scheduler], + [expected, expected, expected], + ); + assert.equal(await client.readiness.check(), true); + }); +} diff --git a/test/back/subscription-mutation-lock.test.cjs b/test/back/subscription-mutation-lock.test.cjs new file mode 100644 index 00000000..fc86d283 --- /dev/null +++ b/test/back/subscription-mutation-lock.test.cjs @@ -0,0 +1,83 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const { spawn } = require('node:child_process'); +const load = require('../helpers/load-security-module.cjs'); + +function fixture(t) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ql-subscription-lock-')); + t.after(() => fs.rmSync(root, { recursive: true, force: true })); + const config = { scriptPath: path.join(root, 'scripts/') }; + const { withSubscriptionMutation } = load( + 'back/shared/subscriptionMutationLock.ts', + { '../config': config }, + ); + return { root, config, withSubscriptionMutation }; +} + +test('subscription ownership mutations exclude a second process until cleanup completes', async (t) => { + const f = fixture(t); + let release, entered; + const gate = new Promise((resolve) => (release = resolve)); + const held = new Promise((resolve) => (entered = resolve)); + t.after(() => release()); + const first = f.withSubscriptionMutation(async () => { + entered(); + await gate; + }); + await held; + const marker = path.join(f.root, 'second-writer'); + const source = ` + const load = require(${JSON.stringify( + path.resolve('test/helpers/load-security-module.cjs'), + )}); + const { withSubscriptionMutation } = load(${JSON.stringify( + path.resolve('back/shared/subscriptionMutationLock.ts'), + )}, { '../config': ${JSON.stringify(f.config)} }); + process.send('attempt'); + withSubscriptionMutation(async () => require('fs').writeFileSync(${JSON.stringify( + marker, + )}, 'done')) + .then(() => process.exit(0), (error) => { console.error(error); process.exit(1); }); + `; + const child = spawn(process.execPath, ['-e', source], { + stdio: ['ignore', 'ignore', 'pipe', 'ipc'], + }); + t.after(() => child.kill()); + let errors = ''; + child.stderr.on('data', (chunk) => (errors += chunk)); + const completed = new Promise((resolve, reject) => { + child.once('error', reject); + child.once('exit', (code) => + code === 0 + ? resolve() + : reject(new Error(errors || `child exit ${code}`)), + ); + }); + await new Promise((resolve) => child.once('message', resolve)); + await new Promise((resolve) => setTimeout(resolve, 150)); + assert.equal(fs.existsSync(marker), false); + release(); + await Promise.all([first, completed]); + assert.equal(fs.readFileSync(marker, 'utf8'), 'done'); + assert.equal( + fs.existsSync(`${path.resolve(f.config.scriptPath)}.subscriptions.lock`), + false, + ); +}); + +test('failed subscription mutations release the lock before the next operation', async (t) => { + const f = fixture(t); + await assert.rejects( + f.withSubscriptionMutation(async () => { + throw new Error('failed cleanup'); + }), + /failed cleanup/, + ); + assert.equal( + await f.withSubscriptionMutation(async () => 'next mutation'), + 'next mutation', + ); +}); diff --git a/test/back/subscription-review-regression.test.cjs b/test/back/subscription-review-regression.test.cjs index 1d3bdece..71b56b9a 100644 --- a/test/back/subscription-review-regression.test.cjs +++ b/test/back/subscription-review-regression.test.cjs @@ -219,7 +219,7 @@ test('startup retains historical aliases and isolates bad rows and asynchronous test('subscription create and update preserve branch aliases without rewriting records', async (t) => { const root = temporary(t); const Subscription = load(path.resolve('back/services/subscription.ts'), { - '../config': { logPath: root }, + '../config': { logPath: root, scriptPath: root }, '../data/subscription': { Subscription: class { constructor(value) { @@ -429,7 +429,7 @@ function cancellationFixture(t, subscriptions) { '../data/subscription': { SubscriptionModel: { findAll: async ({ where }) => - docs.filter((doc) => where.id.includes(doc.id)), + docs.filter((doc) => !where.id || where.id.includes(doc.id)), destroy: async ({ where }) => events.push(['destroy', where.id]), update: async (value, { where }) => { for (const doc of docs) { diff --git a/test/back/subscription-storage-regression.test.cjs b/test/back/subscription-storage-regression.test.cjs new file mode 100644 index 00000000..a27230a1 --- /dev/null +++ b/test/back/subscription-storage-regression.test.cjs @@ -0,0 +1,553 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const os = require('node:os'); +const { execFileSync } = require('node:child_process'); +const load = require('../helpers/load-security-module.cjs'); +const { getSubscriptionStorageNames, getSubscriptionStorageCandidates } = load( + 'back/shared/subscriptionStorage.ts', +); +const { formatUrl } = load('back/config/subscription.ts'); +const cases = [ + { + type: 'public-repo', + url: 'git://127.0.0.1:9418/public.git', + branch: 'main', + }, + { + type: 'public-repo', + url: 'https://github.com/owner/repo.git', + branch: 'feature/release+hotfix', + }, + { type: 'public-repo', url: 'https://github.com/owner/repo.git/' }, + { + type: 'private-repo', + pull_type: 'ssh-key', + url: 'root@127.0.0.1:/qa/fixtures/private.git', + branch: 'main', + }, + { + type: 'private-repo', + pull_type: 'ssh-key', + url: 'git@github.com:owner/repo.git', + branch: 'dev', + }, + { + type: 'private-repo', + pull_type: 'user-pwd', + pull_option: { username: 'tester', password: 'password' }, + url: 'http://127.0.0.1:8788/git/http_private.git', + branch: 'main', + }, + { type: 'file', url: 'http://127.0.0.1:8788/raw.js' }, + { type: 'file', url: 'https://example.com/owner/job.py', branch: 'ignored' }, +].map((doc, index) => ({ + ...doc, + id: index + 1, + alias: `custom/alias+${index}`, +})); +const source = fs.readFileSync('shell/update.sh', 'utf8'); +const bashFunction = source.match(/^get_uniq_path\(\) \{[\s\S]*?^\}/m)[0]; +function shellNames(doc) { + const url = formatUrl(doc).url; + const stem = execFileSync( + 'bash', + [ + '-c', + `${bashFunction}\nget_uniq_path "$1" "$2"\nprintf '%s' "$uniq_path"`, + '--', + url, + doc.type === 'file' ? '' : doc.branch || '', + ], + { encoding: 'utf8' }, + ); + if (doc.type === 'file') { + const raw = `${stem}.${url.slice(url.lastIndexOf('.') + 1)}`; + return { script: `raw_${raw}`, raw }; + } + return { script: stem, repo: stem }; +} + +test('subscription deletion paths match the actual shell URL/branch naming for public, private and raw subscriptions', () => { + for (const doc of cases) + assert.deepEqual( + getSubscriptionStorageNames(doc), + shellNames(doc), + doc.url, + ); +}); + +function fixture(t, values) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ql-storage-')); + t.after(() => fs.rmSync(root, { recursive: true, force: true })); + const config = { + dataPath: root, + scriptPath: path.join(root, 'scripts'), + repoPath: path.join(root, 'repo'), + }; + fs.mkdirSync(config.scriptPath, { recursive: true }); + fs.mkdirSync(config.repoPath, { recursive: true }); + const events = []; + const docs = values.map((value) => ({ + ...value, + get() { + return { ...this }; + }, + })); + const Service = load('back/services/subscription.ts', { + '../config': config, + '../data/subscription': { + Subscription: class { + constructor(value) { + Object.assign(this, value); + } + }, + SubscriptionModel: { + findAll: async ({ where }) => + docs.filter((doc) => !where.id || where.id.includes(doc.id)), + create: async (value) => { + const doc = { + ...value, + get() { + return { ...this }; + }, + }; + docs.push(doc); + return doc; + }, + findOne: async ({ where }) => docs.find((doc) => doc.id === where.id), + update: async (value, { where }) => { + Object.assign( + docs.find((doc) => doc.id === where.id), + value, + ); + }, + destroy: async ({ where }) => { + events.push('destroy'); + for (let i = docs.length - 1; i >= 0; i--) { + if (where.id.includes(docs[i].id)) docs.splice(i, 1); + } + }, + }, + }, + '../data/cron': { CrontabModel: { findAll: async () => [{ id: 99 }] } }, + '../config/util': { + rmPath: (p) => fs.promises.rm(p, { recursive: true, force: true }), + }, + '../config/const': {}, + '../shared/i18n': { t: (x) => x, tf: (x) => x }, + '../shared/pLimit': {}, + './schedule': {}, + './sock': {}, + './sshKey': {}, + './cron': {}, + }).default; + const service = new Service( + {}, + {}, + {}, + { removeSSHKey: async (alias) => events.push(['key', alias]) }, + { remove: async (ids) => events.push(['crons', ids]) }, + ); + service.handleTask = async () => events.push('cancel'); + service.setSshConfig = async () => {}; + const paths = (doc) => + Object.entries(shellNames(doc)).map(([kind, name]) => + path.join( + kind === 'script' + ? config.scriptPath + : kind === 'repo' + ? config.repoPath + : path.join(root, 'raw'), + name, + ), + ); + const write = (target) => { + fs.mkdirSync(path.dirname(target), { recursive: true }); + fs.writeFileSync(target, 'fixture'); + }; + return { root, config, docs, events, service, paths, write }; +} + +test('force deletion cleans URL-derived clone/script/raw artifacts and associated jobs, preserving unrelated alias folders', async (t) => { + const f = fixture(t, cases); + const artifacts = []; + for (const doc of cases) { + for (const dir of f.paths(doc)) { + const artifact = doc.type === 'file' ? dir : path.join(dir, 'job.js'); + artifacts.push(artifact); + f.write(artifact); + } + f.write(path.join(f.config.scriptPath, doc.alias, 'unrelated.js')); + } + await f.service.remove( + cases.map((doc) => doc.id), + { force: true }, + ); + for (const artifact of artifacts) + assert.equal(fs.existsSync(artifact), false, artifact); + for (const doc of cases) + assert.equal( + fs.readFileSync( + path.join(f.config.scriptPath, doc.alias, 'unrelated.js'), + 'utf8', + ), + 'fixture', + ); + assert.ok(f.events.some((e) => Array.isArray(e) && e[0] === 'crons')); + assert.equal( + f.events.filter((e) => Array.isArray(e) && e[0] === 'key').length, + 2, + ); +}); + +test('ordinary deletion preserves subscription artifacts', async (t) => { + const f = fixture(t, [cases[0]]); + const file = path.join(f.paths(cases[0])[0], 'job.js'); + f.write(file); + await f.service.remove([1], {}); + assert.equal(fs.readFileSync(file, 'utf8'), 'fixture'); +}); + +test('a URL-derived symlink escape rejects the entire force deletion before cancelling jobs or deleting rows', async (t) => { + const f = fixture(t, [cases[0], cases[6]]); + const outside = path.join(f.root, 'outside'); + fs.mkdirSync(outside); + fs.writeFileSync(path.join(outside, 'keep'), 'keep'); + fs.mkdirSync(f.config.repoPath, { recursive: true }); + fs.symlinkSync(outside, f.paths(cases[0])[1]); + await assert.rejects( + f.service.remove([1, 7], { force: true }), + /Invalid subscription path/, + ); + assert.deepEqual(f.events, []); + assert.equal(fs.readFileSync(path.join(outside, 'keep'), 'utf8'), 'keep'); +}); + +const cliCases = [ + { + type: 'public-repo', + url: 'https://github.com/owner/repo.git/', + branch: 'main', + }, + { + type: 'private-repo', + url: 'git@github.com:repo.git', + branch: 'main', + pull_type: 'ssh-key', + }, + { type: 'file', url: 'https://example.com/owner/job.py?version=2' }, + { type: 'file', url: 'https://example.com/owner/job' }, +].map((doc, index) => ({ ...doc, id: index + 1, alias: `cli-${index}` })); + +test('cleanup candidates match both actual CLI naming and legacy Shell naming', () => { + const { repositoryName } = load( + 'cli/src/internal/subscription/subscriptionRunner.ts', + ); + for (const doc of [...cases, ...cliCases]) { + const url = formatUrl(doc).url; + const stem = repositoryName( + url, + doc.type === 'file' ? undefined : doc.branch, + ); + const raw = + doc.type === 'file' + ? stem + path.extname(new URL(url).pathname) + : undefined; + assert.deepEqual( + getSubscriptionStorageNames(doc, 'cli'), + raw === undefined + ? { script: stem, repo: stem } + : { script: `raw_${raw}`, raw }, + ); + assert.deepEqual(getSubscriptionStorageNames(doc), shellNames(doc)); + } +}); + +test('force deletion removes CLI and legacy artifacts for trailing slashes, SCP URLs and raw URL queries', async (t) => { + const f = fixture(t, cliCases); + const artifacts = []; + for (const doc of cliCases) + for (const names of getSubscriptionStorageCandidates(doc)) { + for (const [kind, name] of Object.entries(names)) { + const base = + kind === 'script' + ? f.config.scriptPath + : kind === 'repo' + ? f.config.repoPath + : path.join(f.root, 'raw'); + const target = path.join( + base, + name, + ...(doc.type === 'file' ? [] : ['job.js']), + ); + artifacts.push(target); + f.write(target); + } + } + await f.service.remove( + cliCases.map((doc) => doc.id), + { force: true }, + ); + for (const file of artifacts) assert.equal(fs.existsSync(file), false, file); +}); + +for (const type of ['public-repo', 'file']) { + test(`force deletion retains shared ${type} artifacts until the last subscription is removed`, async (t) => { + const first = { ...cases[type === 'file' ? 6 : 0], id: 1, alias: 'first' }; + const second = { ...first, id: 2, alias: 'second' }; + const f = fixture(t, [first, second]); + const artifacts = f + .paths(first) + .map((p) => (type === 'file' ? p : path.join(p, 'job.js'))); + artifacts.forEach(f.write); + await f.service.remove([1], { force: true }); + for (const file of artifacts) assert.equal(fs.existsSync(file), true, file); + await f.service.remove([2], { force: true }); + for (const file of artifacts) + assert.equal(fs.existsSync(file), false, file); + }); +} + +test('batch deletion of every reference removes shared storage once', async (t) => { + const f = fixture(t, [ + { ...cases[0], id: 1 }, + { ...cases[0], id: 2, alias: 'second' }, + ]); + const artifacts = f.paths(cases[0]).map((p) => path.join(p, 'job.js')); + artifacts.forEach(f.write); + await f.service.remove([1, 2], { force: true }); + for (const file of artifacts) assert.equal(fs.existsSync(file), false, file); +}); + +test('recursive cleanup retains overlapping parent and nested branch directories', async (t) => { + const parent = { ...cases[0], id: 1, branch: 'main', alias: 'parent' }; + const child = { ...parent, id: 2, branch: 'main/nested', alias: 'child' }; + for (const removed of [1, 2]) { + const f = fixture(t, [parent, child]); + const artifacts = [parent, child].flatMap((doc) => + f.paths(doc).map((p) => path.join(p, 'job.js')), + ); + artifacts.forEach(f.write); + await f.service.remove([removed], { force: true }); + for (const file of artifacts) assert.equal(fs.existsSync(file), true, file); + } +}); + +test('unrelated malformed historical rows do not block force deletion, and malformed shared references still retain storage', async (t) => { + const selected = { ...cases[1], id: 1, branch: 'main', alias: 'selected' }; + for (const shared of [false, true]) { + const invalid = { + id: 2, + alias: '../historical', + type: 'private-repo', + pull_type: 'ssh-key', + url: `git@github.com:${shared ? 'owner/repo' : 'other/unrelated'}.git`, + branch: 'main', + }; + const f = fixture(t, [selected, invalid]); + const files = f.paths(selected).map((p) => path.join(p, 'job.js')); + files.forEach(f.write); + await f.service.remove([1], { force: true }); + assert.deepEqual( + f.docs.map((doc) => doc.id), + [2], + ); + for (const file of files) assert.equal(fs.existsSync(file), shared); + } +}); + +test('an unused invalid Shell candidate does not block cleanup of the valid CLI raw files', async (t) => { + const doc = { + id: 1, + alias: 'valid-cli-raw', + type: 'file', + url: 'https://example.com/job.js?source=https://foo/bar', + }; + const f = fixture(t, [doc]); + const names = getSubscriptionStorageNames(doc, 'cli'); + const files = [ + path.join(f.config.scriptPath, names.script), + path.join(f.root, 'raw', names.raw), + ]; + files.forEach(f.write); + await f.service.remove([1], { force: true }); + assert.deepEqual(f.docs, []); + for (const file of files) assert.equal(fs.existsSync(file), false); +}); + +for (const mutation of ['create', 'update']) { + test(`force deletion prevents a concurrent ${mutation} from changing ownership after its reference snapshot`, async (t) => { + const selected = { ...cases[0], id: 1, alias: 'selected' }; + const other = { + ...selected, + id: 2, + alias: 'other', + url: 'https://github.com/other/unrelated.git', + }; + const f = fixture( + t, + mutation === 'update' ? [selected, other] : [selected], + ); + let release, entered; + const gate = new Promise((resolve) => (release = resolve)); + const held = new Promise((resolve) => (entered = resolve)); + t.after(() => release()); + f.service.handleTask = async (doc, needCreate) => { + if (doc.id === 1 && needCreate === false) { + entered(); + await gate; + } + }; + const deletion = f.service.remove([1], { force: true }); + await held; + const change = f.service[mutation]({ ...selected, id: 2, alias: 'other' }); + await new Promise((resolve) => setTimeout(resolve, 150)); + if (mutation === 'create') + assert.deepEqual( + f.docs.map((doc) => doc.id), + [1], + ); + else assert.equal(f.docs.find((doc) => doc.id === 2).url, other.url); + release(); + await Promise.all([deletion, change]); + assert.equal(f.docs.find((doc) => doc.id === 2).url, selected.url); + }); + + test(`force deletion observes shared storage ownership established by an in-flight ${mutation}`, async (t) => { + const selected = { ...cases[0], id: 1, alias: 'selected' }; + const other = { + ...selected, + id: 2, + alias: 'other', + url: 'https://github.com/other/unrelated.git', + }; + const f = fixture( + t, + mutation === 'update' ? [selected, other] : [selected], + ); + const file = path.join(f.paths(selected)[0], 'job.js'); + f.write(file); + let release, entered; + const gate = new Promise((resolve) => (release = resolve)); + const held = new Promise((resolve) => (entered = resolve)); + t.after(() => release()); + f.service.handleTask = async (doc, needCreate) => { + if (doc.id === 2 && needCreate !== false) { + entered(); + await gate; + } + }; + const change = f.service[mutation]({ ...selected, id: 2, alias: 'other' }); + await held; + const deletion = f.service.remove([1], { force: true }); + await new Promise((resolve) => setTimeout(resolve, 150)); + assert.ok( + f.docs.some((doc) => doc.id === 1), + 'deletion waits until the ownership mutation is complete', + ); + release(); + await Promise.all([change, deletion]); + assert.deepEqual( + f.docs.map((doc) => doc.id), + [2], + ); + assert.equal(fs.existsSync(file), true); + }); +} + +test('an unrelated dangling or outside symlink reference does not block safe cleanup', async (t) => { + const selected = { ...cases[0], id: 1, alias: 'selected' }; + const other = { + ...selected, + id: 2, + alias: 'other', + url: 'https://github.com/other/repo.git', + }; + const f = fixture(t, [selected, other]); + const outside = path.join(f.root, 'outside'); + fs.mkdirSync(outside); + fs.writeFileSync(path.join(outside, 'keep'), 'keep'); + fs.symlinkSync(outside, f.paths(other)[0]); + const file = path.join(f.paths(selected)[0], 'job.js'); + f.write(file); + await f.service.remove([1], { force: true }); + assert.equal(fs.existsSync(file), false); + assert.equal(fs.readFileSync(path.join(outside, 'keep'), 'utf8'), 'keep'); +}); + +test('URL-derived storage accepts legal directory names longer than user aliases and containing spaces', async (t) => { + const doc = { ...cases[0], id: 1, alias: 'short-alias', url: `/tmp/owner/${'r'.repeat(205)} repo.git` }; + const f = fixture(t, [doc]); + const files = f.paths(doc).map((p) => path.join(p, 'job.js')); + files.forEach(f.write); + await f.service.remove([1], { force: true }); + for (const file of files) assert.equal(fs.existsSync(file), false, file); +}); + +test('an unrepresentable unused Shell filename does not block valid CLI raw cleanup', async (t) => { + const doc = { + id: 1, + alias: 'long-query', + type: 'file', + url: 'https://example.com/job.js?token=' + 'a'.repeat(300), + }; + const f = fixture(t, [doc]); + const names = getSubscriptionStorageNames(doc, 'cli'); + const files = [ + path.join(f.config.scriptPath, names.script), + path.join(f.root, 'raw', names.raw), + ]; + files.forEach(f.write); + await f.service.remove([1], { force: true }); + assert.deepEqual(f.docs, []); + for (const file of files) assert.equal(fs.existsSync(file), false); +}); + +test('cleanup retains each representable path when only its prefixed Shell script name is too long', async (t) => { + const doc = { id: 1, alias: 'partial-raw', type: 'file', url: 'https://example.com/job.js?token=' }; + const prefix = getSubscriptionStorageNames(doc).raw; + doc.url += 'a'.repeat(255 - Buffer.byteLength(prefix)); + const shell = getSubscriptionStorageNames(doc); + assert.equal(Buffer.byteLength(shell.raw), 255); + assert.ok(Buffer.byteLength(shell.script) > 255); + const f = fixture(t, [doc]); + const cli = getSubscriptionStorageNames(doc, 'cli'); + const files = [ + path.join(f.root, 'raw', shell.raw), + path.join(f.root, 'raw', cli.raw), + path.join(f.config.scriptPath, cli.script), + ]; + files.forEach(f.write); + await f.service.remove([1], { force: true }); + assert.deepEqual(f.docs, []); + for (const file of files) assert.equal(fs.existsSync(file), false); +}); + +test('a legal Unicode alias with an unrepresentable unused directory does not block URL-derived cleanup', async (t) => { + const doc = { ...cases[0], id: 1, alias: '中'.repeat(86) }; + const f = fixture(t, [doc]); + const files = f.paths(doc).map((p) => path.join(p, 'job.js')); + files.forEach(f.write); + await f.service.remove([1], { force: true }); + assert.deepEqual(f.docs, []); + for (const file of files) assert.equal(fs.existsSync(file), false); +}); + +test('Unicode storage follows filesystem limits rather than imposing a UTF-8 byte cap', async (t) => { + const doc = { ...cases[0], id: 1, alias: 'short-alias', url: `/tmp/owner/${'中'.repeat(86)}.git` }; + const f = fixture(t, [doc]); + const files = f.paths(doc).map((p) => path.join(p, 'job.js')); + for (const file of files) { + try { + f.write(file); + } catch (error) { + // These names are representable on macOS but not Linux. + assert.equal(error.code, 'ENAMETOOLONG'); + } + } + await f.service.remove([1], { force: true }); + assert.deepEqual(f.docs, []); + for (const file of files) assert.equal(fs.existsSync(file), false); +}); diff --git a/test/back/system-dependence-cache.test.cjs b/test/back/system-dependence-cache.test.cjs index c06dd3fb..1df6006d 100644 --- a/test/back/system-dependence-cache.test.cjs +++ b/test/back/system-dependence-cache.test.cjs @@ -59,6 +59,7 @@ test('clearing a dependency cache marks installed entries for reinstall', async AuthDataType: {}, SystemModel: {}, }); + stubModule('../../back/schedule/client', { __esModule: true, default: {} }); stubModule('../../back/shared/pLimit', { __esModule: true, default: {}, diff --git a/test/back/system-export-history.test.cjs b/test/back/system-export-history.test.cjs new file mode 100644 index 00000000..4994a8cb --- /dev/null +++ b/test/back/system-export-history.test.cjs @@ -0,0 +1,119 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs/promises'); +const path = require('node:path'); +const os = require('node:os'); +const { execFileSync } = require('node:child_process'); +const load = require('../helpers/load-security-module.cjs'); +const { ScriptHistory } = load('back/shared/scriptHistory.ts', { + '../config/util': {}, +}); + +async function fixture(t) { + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'ql-history-export-')); + t.after(() => fs.rm(root, { recursive: true, force: true })); + const dataPath = path.join(root, 'data'); + for (const dir of ['db', 'upload', 'scripts']) + await fs.mkdir(path.join(dataPath, dir), { recursive: true }); + const archive = path.join(root, 'backup.tgz'); + const Service = load('back/services/system.ts', { + '../config': { dataPath, dataTgzFile: archive }, + '../config/const': {}, + '../config/util': {}, + '../data/dependence': {}, + '../data/notify': {}, + '../data/system': {}, + '../shared/pLimit': {}, + '../schedule/client': {}, + '../shared/i18n': { t: (x) => x }, + './notify': class {}, + './schedule': class {}, + './sock': class {}, + }).default; + const service = new Service({}, {}, {}); + let downloaded, status, error; + const res = { + download: (file) => { + downloaded = file; + }, + status: (code) => { + status = code; + return res; + }, + send: (value) => { + error = value; + return res; + }, + }; + const exportData = async (types) => { + downloaded = status = error = undefined; + await service.exportData(res, types); + return { downloaded, status, error }; + }; + return { root, dataPath, archive, exportData }; +} + +test('a scripts backup preserves history versions and they remain readable and restorable after extraction', async (t) => { + const f = await fixture(t); + const scripts = path.join(f.dataPath, 'scripts'); + const historyRoot = path.join(f.dataPath, 'script-history'); + await fs.writeFile(path.join(scripts, 'history.js'), 'baseline'); + const history = new ScriptHistory(scripts, historyRoot); + await history.save('', 'history.js', 'second'); + await history.save('', 'history.js', 'third'); + const before = await history.list('', 'history.js'); + assert.ok(before.versions.length >= 2); + assert.deepEqual(await f.exportData(['scripts']), { + downloaded: f.archive, + status: undefined, + error: undefined, + }); + const restored = path.join(f.root, 'restored'); + await fs.mkdir(restored); + execFileSync('tar', ['-zxf', f.archive, '-C', restored]); + const after = new ScriptHistory( + path.join(restored, 'data/scripts'), + path.join(restored, 'data/script-history'), + ); + assert.deepEqual(await after.list('', 'history.js'), before); + const oldest = before.versions.at(-1); + const detail = await after.detail('', 'history.js', oldest.id); + assert.equal(detail.version.content, 'baseline'); + await after.restore('', 'history.js', oldest.id, before.currentHash); + assert.equal( + await fs.readFile(path.join(restored, 'data/scripts/history.js'), 'utf8'), + 'baseline', + ); +}); + +test('fresh installations export scripts without requiring a history directory', async (t) => { + const f = await fixture(t); + await fs.writeFile(path.join(f.dataPath, 'scripts/job.js'), 'fresh'); + assert.equal((await f.exportData(['scripts'])).downloaded, f.archive); + const entries = execFileSync('tar', ['-ztf', f.archive], { + encoding: 'utf8', + }); + assert.match(entries, /data\/scripts\/job.js/); + assert.doesNotMatch(entries, /script-history/); +}); + +test('history can be exported explicitly, while directory traversal is rejected before tar runs', async (t) => { + const f = await fixture(t); + await fs.mkdir(path.join(f.dataPath, 'script-history')); + await fs.writeFile( + path.join(f.dataPath, 'script-history/version.br'), + 'history', + ); + assert.equal((await f.exportData(['script-history'])).downloaded, f.archive); + assert.match( + execFileSync('tar', ['-ztf', f.archive], { encoding: 'utf8' }), + /script-history\/version.br/, + ); + await fs.rm(f.archive); + for (const type of ['../outside', '/etc', 'scripts/../../outside']) { + const result = await f.exportData([type]); + assert.equal(result.status, 400); + assert.equal(result.downloaded, undefined); + } + await assert.rejects(fs.stat(f.archive), { code: 'ENOENT' }); +}); diff --git a/test/back/wpush-grpc.test.cjs b/test/back/wpush-grpc.test.cjs index c75679af..3229882d 100644 --- a/test/back/wpush-grpc.test.cjs +++ b/test/back/wpush-grpc.test.cjs @@ -66,6 +66,7 @@ test('decoded SystemNotify requests reach WPUSH with explicit or saved configura '../data/dependence': {}, '../data/system': {}, '../shared/pLimit': {}, + '../schedule/client': {}, '../shared/i18n': { t: (x) => x }, './notify': {}, './schedule': {},