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': {},