Compare commits

...
Author SHA1 Message Date
whyour 7603702eca fix: reduce task startup overhead and lazy-load notifications 2026-09-28 09:42:54 +08:00
whyour f168efcdae refactor: migrate cron scheduling to node-cron (#3079)
* refactor: migrate cron scheduling to node-cron

* feat: support annually midnight and minutely cron macros

* fix: harden cron recovery and system scheduler compatibility
2026-09-27 22:18:39 +08:00
27 changed files with 2232 additions and 209 deletions
+3 -10
View File
@@ -3,7 +3,7 @@ import { Container } from 'typedi';
import { Logger } from 'winston';
import SubscriptionService from '../services/subscription';
import { celebrate, Joi } from 'celebrate';
import CronExpressionParser from 'cron-parser';
import { isValidCronSchedule } from '../shared/cronSchedule';
const route = Router();
export default (app: Router) => {
@@ -58,10 +58,7 @@ export default (app: Router) => {
async (req: Request, res: Response, next: NextFunction) => {
const logger: Logger = Container.get('logger');
try {
if (
!req.body.schedule ||
CronExpressionParser.parse(req.body.schedule).hasNext()
) {
if (!req.body.schedule || isValidCronSchedule(req.body.schedule)) {
const subscriptionService = Container.get(SubscriptionService);
const data = await subscriptionService.create(req.body);
return res.send({ code: 200, data });
@@ -204,11 +201,7 @@ export default (app: Router) => {
async (req: Request, res: Response, next: NextFunction) => {
const logger: Logger = Container.get('logger');
try {
if (
!req.body.schedule ||
typeof req.body.schedule === 'object' ||
CronExpressionParser.parse(req.body.schedule).hasNext()
) {
if (!req.body.schedule || isValidCronSchedule(req.body.schedule)) {
const subscriptionService = Container.get(SubscriptionService);
const data = await subscriptionService.update(req.body);
return res.send({ code: 200, data });
+48 -73
View File
@@ -1,6 +1,6 @@
import { ServerUnaryCall, sendUnaryData, status } from '@grpc/grpc-js';
import { AddCronRequest, AddCronResponse } from '../protos/cron';
import nodeSchedule from 'node-schedule';
import { createCronJob, CronJob } from '../shared/cronScheduler';
import { isValidCronSchedule } from '../shared/cronSchedule';
import { scheduleStacks } from './data';
import { runCron } from '../shared/runCron';
@@ -24,11 +24,7 @@ const addCron = (
if (!isValidCronField(schedule)) {
validationErrors.push(
tf(
'任务ID %s: 无效的 cron 表达式 "%s"',
String(id),
schedule,
),
tf('任务ID %s: 无效的 cron 表达式 "%s"', String(id), schedule),
);
}
@@ -56,79 +52,58 @@ const addCron = (
return;
}
// Recovery replaces the whole snapshot, including deletions and disabled jobs.
// Validation above must finish before touching the previous schedule.
if (call.request.replace) {
for (const jobs of scheduleStacks.values()) {
for (const job of jobs) job?.cancel();
// Prepare stopped jobs first: construction errors must preserve the old snapshot.
const prepared = new Map<string, CronJob[]>();
try {
for (const item of call.request.crons) {
const jobs: CronJob[] = [];
prepared.get(item.id)?.forEach((job) => job.cancel());
prepared.set(item.id, jobs);
for (const schedule of [
item.schedule,
...(item.extra_schedules || []).map((x) => x.schedule),
]) {
jobs.push(
createCronJob(
schedule,
async () => {
Logger.info('[schedule][准备运行任务] 命令: %s', item.command);
await runCron(item.command, item);
},
{
name: `${item.id}: ${item.name || ''}`,
logger: Logger,
start: false,
},
),
);
}
}
scheduleStacks.clear();
} catch (error) {
for (const jobs of prepared.values()) jobs.forEach((job) => job.cancel());
const err: any = new Error(
error instanceof Error ? error.message : String(error),
);
err.code = status.INVALID_ARGUMENT;
err.details = err.message;
callback(err, null);
return;
}
// ===== 第二遍:注册所有任务 =====
for (const item of call.request.crons) {
const { id, schedule, command, extra_schedules, name } = item;
// 取消该 id 已有的旧任务
if (scheduleStacks.has(id)) {
scheduleStacks.get(id)?.forEach((x) => x.cancel());
}
if (call.request.replace) {
for (const jobs of scheduleStacks.values())
jobs.forEach((job) => job?.cancel());
scheduleStacks.clear();
}
for (const [id, jobs] of prepared) {
scheduleStacks.get(id)?.forEach((job) => job.cancel());
scheduleStacks.set(id, jobs);
jobs.forEach((job) => job.start());
Logger.info(
'[schedule][创建定时任务] 任务ID: %s, 名称: %s, cron: %s, 执行命令: %s',
'[schedule][创建定时任务] 任务ID: %s, 规则数: %s',
id,
name,
schedule,
command,
jobs.length,
);
if (extra_schedules?.length) {
extra_schedules.forEach((x) => {
Logger.info(
'[schedule][创建定时任务] 任务ID: %s, 名称: %s, cron: %s, 执行命令: %s',
id,
name,
x.schedule,
command,
);
});
}
const mainJob = nodeSchedule.scheduleJob(id, schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
});
if (!mainJob) {
Logger.warn(
'[schedule][创建定时任务] scheduleJob 返回 null(不符合预期,已通过预校验): 任务ID: %s, cron: %s',
id,
schedule,
);
}
const extraJobs = extra_schedules?.length
? extra_schedules.map((x) => {
const job = nodeSchedule.scheduleJob(id, x.schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
});
if (!job) {
Logger.warn(
'[schedule][创建定时任务] scheduleJob 返回 null(不符合预期,已通过预校验): 任务ID: %s, cron: %s',
id,
x.schedule,
);
}
return job;
})
: [];
// 过滤 null(兜底保护,正常情况下预校验已拦截)
const jobs = [mainJob, ...extraJobs].filter((x) => x != null);
if (jobs.length > 0) {
scheduleStacks.set(id, jobs);
}
}
callback(null, null);
+7 -1
View File
@@ -67,11 +67,17 @@ class Client {
replace = false
): Promise<AddCronResponse> {
await this.waitForReady(2000);
// Include every rule in a recovery snapshot, but bound a stuck worker.
const ruleCount = request.reduce(
(count, item) => count + 1 + (item.extra_schedules?.length || 0),
0,
);
const registrationTimeoutMs = Math.min(120000, 5000 + ruleCount * 5);
return new Promise((resolve, reject) => {
this.client.addCron(
{ crons: request, replace },
new Metadata(),
{ deadline: Date.now() + 5000 },
{ deadline: Date.now() + registrationTimeoutMs },
(err, res) => {
if (err) {
if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) {
+2 -2
View File
@@ -1,6 +1,6 @@
import nodeSchedule from 'node-schedule';
import { CronJob } from '../shared/cronScheduler';
import { ToadScheduler } from 'toad-scheduler';
export const scheduleStacks = new Map<string, nodeSchedule.Job[]>();
export const scheduleStacks = new Map<string, CronJob[]>();
export const intervalSchedule = new ToadScheduler();
+1 -1
View File
@@ -13,7 +13,7 @@ const delCron = (
'[schedule][取消定时任务] 任务ID: %s',
id,
);
// 过滤掉 nodeSchedule.scheduleJob() 对无效表达式返回的 null,
// 防御性过滤历史调度栈中的空任务,
// 否则对 null 调 cancel() 会让整个取消流程抛出 UNKNOWN 错误,
// 进而导致 HTTP 端的 remove() 跳过 setCrontab(),造成 crontab.list 残留。
scheduleStacks.get(id)?.filter((x) => x != null).forEach((x) => {
+13 -6
View File
@@ -1,5 +1,5 @@
import { randomUUID } from 'crypto';
import { getInvalidCronSchedules } from '../shared/cronSchedule';
import { getInvalidCronSchedules, isValidCronSchedule } from '../shared/cronSchedule';
import {
withSchedulerMutation,
schedulerRegistrationError,
@@ -14,7 +14,6 @@ import {
} from '../data/runningInstance';
import { exec, execSync } from 'child_process';
import fs from 'fs/promises';
import CronExpressionParser from 'cron-parser';
import {
getFileContentByName,
fileExist,
@@ -47,12 +46,20 @@ export default class CronService {
private isNodeCron(cron: Crontab) {
const { schedule, extra_schedules } = cron;
const fields = schedule?.trim().split(/\s+/) || [];
// System crontab only receives portable numeric five-field expressions.
// Extended syntax, macros and legacy shorthand belong to node-schedule.
// Extended syntax, macros and legacy shorthand use the Node scheduler.
return (
schedule?.trim().split(/\s+/).length !== 5 ||
fields.length !== 5 ||
/[^\d\s*,/\-]/.test(schedule || '') ||
Boolean(extra_schedules?.length)
Boolean(extra_schedules?.length) ||
// BusyBox treats N/step as a single value and doesn't support Sunday=7.
fields.some((field) => field.split(',').some((part) => /^\d+\//.test(part))) ||
fields[4].includes('7') ||
// BusyBox steps DOM from zero; its full-range day/week wildcard
// handling also differs from the legacy Node parser's OR semantics.
fields[2].includes('/') ||
(fields[2] !== '*' && fields[4] !== '*')
);
}
@@ -1057,7 +1064,7 @@ export default class CronService {
if (
command &&
schedule &&
CronExpressionParser.parse(schedule).hasNext()
isValidCronSchedule(schedule)
) {
const name = namePrefix + '_' + index;
+33 -14
View File
@@ -1,6 +1,6 @@
import { Service, Inject } from 'typedi';
import winston from 'winston';
import nodeSchedule from 'node-schedule';
import { createCronJob, CronJob } from '../shared/cronScheduler';
import { ChildProcessWithoutNullStreams } from 'child_process';
import {
ToadScheduler,
@@ -11,7 +11,11 @@ import {
import dayjs from 'dayjs';
import taskLimit from '../shared/pLimit';
import { spawn } from 'cross-spawn';
import { observeChildProcess, asError, ProcessResult } from '../shared/childProcess';
import {
observeChildProcess,
asError,
ProcessResult,
} from '../shared/childProcess';
export interface ScheduleTaskType {
id?: number;
@@ -38,7 +42,7 @@ export interface TaskCallbacks {
@Service()
export default class ScheduleService {
private scheduleStacks = new Map<string, nodeSchedule.Job>();
private scheduleStacks = new Map<string, CronJob>();
private intervalSchedule = new ToadScheduler();
@@ -159,18 +163,33 @@ export default class ScheduleService {
command,
);
this.scheduleStacks.set(
_id,
nodeSchedule.scheduleJob(_id, schedule, async () => {
this.runTask(command, callbacks, {
name,
if (schedule) {
try {
const job = createCronJob(
schedule,
command,
id: _id,
runOrigin,
});
}),
);
() =>
this.runTask(command, callbacks, {
name,
schedule,
command,
id: _id,
runOrigin,
}),
{ name: `${_id}: ${name || ''}`, logger: this.logger, start: false },
);
this.scheduleStacks.get(_id)?.cancel();
this.scheduleStacks.set(_id, job);
job.start();
} catch (error) {
// A persisted invalid subscription must not become an unhandled startup rejection.
this.logger.warn(
'[panel][跳过无效定时任务] 任务ID: %s, cron: %s, 错误: %s',
_id,
schedule,
asError(error).message,
);
}
}
if (runImmediately) {
this.runTask(command, callbacks, {
+90 -7
View File
@@ -1,12 +1,93 @@
import cronParser from 'cron-parser-v4';
import CronExpressionParser from 'cron-parser';
import { validate } from 'node-cron';
import { ScheduleType } from '../interface/schedule';
// Keep validation aligned with node-schedule's locked cron-parser version.
// cron-parser 5 accepts syntax (e.g. H) that the scheduler cannot execute.
const aliases: Record<string, string> = {
'@yearly': '0 0 0 1 1 *',
'@annually': '0 0 0 1 1 *',
'@monthly': '0 0 0 1 * *',
'@weekly': '0 0 0 * * 0',
'@daily': '0 0 0 * * *',
'@midnight': '0 0 0 * * *',
'@hourly': '0 0 * * * *',
'@minutely': '0 * * * * *',
};
export interface CronSchedule {
patterns: string[];
dayCount: number;
nthDay: number;
}
// Validation and registration repeatedly see identical rules in large snapshots.
// Cache only immutable syntax, never dates or running tasks, and bound user input.
const scheduleCache = new Map<string, CronSchedule>();
// Canonicalize legacy shorthand, aliases, names and numeric-start steps before
// node-cron sees them. Extend legacy aliases explicitly; reject H and bare /N.
export function parseCronSchedule(schedule: unknown): CronSchedule {
if (typeof schedule !== 'string' || !schedule.trim()) {
throw new Error('Invalid cron schedule');
}
let source = schedule.trim();
const cacheKey = source;
const cached = scheduleCache.get(cacheKey);
if (cached) return cached;
if (source.startsWith('@')) {
if (!Object.hasOwn(aliases, source)) throw new Error('Invalid cron alias');
source = aliases[source];
}
let parts = source.split(/\s+/);
if (
parts.length > 6 ||
parts.some((part) => /(^|,)\//.test(part) || /H(?:\(|\/|$)/i.test(part))
) {
throw new Error('Unsupported cron syntax');
}
parts = [
...['0', '*', '*', '*', '*', '*'].slice(0, 6 - parts.length),
...parts,
];
if (
parts.some(
(part, index) => index !== 3 && index !== 5 && part.includes('?'),
)
) {
throw new Error('Question mark is only valid in day fields');
}
const expression = CronExpressionParser.parse(parts.join(' '));
if (!expression.hasNext())
throw new Error('Cron schedule has no next execution');
const normalized = expression.stringify(true).replace(/\?/g, '*').split(' ');
const dayCount = expression.fields.dayOfMonth.values.length;
const weekCount = expression.fields.dayOfWeek.values.length;
const patterns =
dayCount < 31 && weekCount < 8
? [
[...normalized.slice(0, 5), '*'].join(' '),
[...normalized.slice(0, 3), '*', ...normalized.slice(4)].join(' '),
]
: [normalized.join(' ')];
if (!patterns.every((pattern) => validate(pattern))) {
throw new Error('Unsupported cron schedule');
}
const parsed = {
patterns,
dayCount,
nthDay: expression.fields.dayOfWeek.nthDay,
};
Object.freeze(patterns);
Object.freeze(parsed);
if (scheduleCache.size >= 512)
scheduleCache.delete(scheduleCache.keys().next().value);
scheduleCache.set(cacheKey, parsed);
return parsed;
}
export function isValidCronSchedule(schedule: unknown): schedule is string {
if (typeof schedule !== 'string' || !schedule.trim()) return false;
try {
return cronParser.parseExpression(schedule).hasNext();
parseCronSchedule(schedule);
return true;
} catch {
return false;
}
@@ -22,6 +103,8 @@ export function getInvalidCronSchedules(cron: {
) {
return [];
}
return [cron.schedule, ...(cron.extra_schedules || []).map((x) => x.schedule)]
.filter((schedule) => !isValidCronSchedule(schedule));
return [
cron.schedule,
...(cron.extra_schedules || []).map((x) => x.schedule),
].filter((schedule) => !isValidCronSchedule(schedule));
}
+89
View File
@@ -0,0 +1,89 @@
import { createTask, ScheduledTask } from 'node-cron';
import { parseCronSchedule } from './cronSchedule';
interface SchedulerLogger {
warn(message: string, ...args: unknown[]): unknown;
error(message: string, ...args: unknown[]): unknown;
}
export interface CronJob {
start(): void;
cancel(): void;
}
// node-cron owns timers and calendar calculation; Qinglong owns execution
// concurrency and process lifetime. Missed in-process slots join that same queue.
export function createCronJob(
schedule: string,
callback: (date: Date) => unknown | Promise<unknown>,
options: { name: string; logger: SchedulerLogger; start?: boolean },
): CronJob {
const parsed = parseCronSchedule(schedule);
const tasks: ScheduledTask[] = [];
let cancelled = false;
let started = false;
const accepts = (index: number, date: Date) => {
if (tasks.length === 1) return true;
// Preserve cron-parser 4's day/weekday OR semantics, including its
// month-dependent full-day range and the global nth-week constraint.
if (parsed.nthDay && Math.ceil(date.getDate() / 7) !== parsed.nthDay)
return false;
if (index === 1) return !tasks[0].match(date);
const monthDays = [31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
return parsed.dayCount < monthDays[date.getMonth()] || tasks[1].match(date);
};
const dispatch = async (index: number, date: Date, missed: boolean) => {
if (cancelled || !started || !accepts(index, date)) return;
if (missed) {
options.logger.warn(
'[schedule][补执行迟到任务] 任务: %s, 计划时间: %s',
options.name,
date.toISOString(),
);
}
try {
await callback(date);
} catch (error) {
options.logger.error(
'[schedule][定时回调失败] 任务: %s, 错误: %s',
options.name,
error instanceof Error ? error.message : String(error),
);
}
};
try {
for (const [index, pattern] of parsed.patterns.entries()) {
const task = createTask(
pattern,
(context) => dispatch(index, context.date, false),
{
noOverlap: false,
missedExecutionTolerance: 5000,
},
);
tasks.push(task);
task.on('execution:missed', (context) =>
dispatch(index, context.date, true),
);
// Fail before an existing schedule is cancelled, including impossible dates.
task.getNextRuns(1);
}
} catch (error) {
tasks.forEach((task) => task.destroy());
throw error;
}
const job: CronJob = {
start() {
if (started || cancelled) return;
started = true;
tasks.forEach((task) => task.start());
},
cancel() {
if (cancelled) return;
cancelled = true;
tasks.forEach((task) => task.destroy());
},
};
if (options.start !== false) job.start();
return job;
}
+49
View File
@@ -0,0 +1,49 @@
# Node cron 调度器迁移
Node 调度路径使用固定版本 `node-cron@4.6.0`,普通任务(含额外规则)、订阅和系统 cron 共用 `back/shared/cronScheduler.ts`。启动/一次性任务、间隔任务、脚本并发限制和进程执行链路保持原有实现。系统模式只将语义一致的普通五字段规则交给 crond。
移除 `node-schedule`、其类型包以及兼容别名 `cron-parser-v4`。保留 `cron-parser@5` 做旧表达式的校验与标准化;计时器和日历推进由 node-cron 管理。
## 兼容行为
- 保留旧版 1–6 字段补全、五种宏、月份/星期名称、数字起点步长、L 和 #。另外明确新增 `@annually`(等同 `@yearly`)、`@midnight`(等同 `@daily`)、`@minutely`(每分钟第 0 秒)。宏按进程时区解释。`?` 仅允许日期和星期字段;裸 `/N`、H 和其他未列出的宏仍拒绝。
- 日期与星期同时指定时保留 OR 语义,不采用 node-cron 默认的 AND。两个原生任务分别处理日期和星期,并按计划时间过滤交集,避免重复执行;保留旧解析器的月份全日期范围及全局 nth-week 限制。
- 原生任务先暂停创建并验证,再替换旧任务。构造失败时销毁已准备的实例并保留旧快照。
- 进程内迟到的触发通过 `execution:missed` 加入既有执行队列,记录补执行日志。并发和单任务积压上限仍由青龙队列控制;这不提供进程退出期间的持久化补跑或分布式 exactly-once 保证。
- 标准化结果使用上限 512 条的只读缓存,仅缓存语法,不缓存运行实例或下次执行时间。避免大批相同表达式重复解析导致恢复 RPC 超时。
- 回调异常记录日志,后续调度继续;取消时调用原生 `destroy()` 释放计时器和注册表。
- 持久化无效任务在恢复时跳过并记录日志,无效订阅不会产生未处理的启动拒绝。API、导入和执行使用同一套表达式校验。
## 复审修复
- 移除没有生产调用的 `nextInvocation()` 接口,避免对 OR/# 等过滤规则报告不会实际执行的原生日期。回归测试直接观察实际执行和原生实例状态。
- 批量注册 RPC 时限为 5 秒加每条规则 5 毫秒,计入额外规则并封顶 120 秒;保留超时后的恢复机制,不立即重放结果不确定的写入。
- Alpine/系统模式下,数字起点步长(如 `0/5`)、星期包含 `7`、日期步长、日期和星期同时受限的表达式转入 Node 路径。BusyBox 的这些解析或通配语义与旧 Node 解析器不同,不能仅因是纯数字五字段就交给 crond。
- 这些规则在 `crontab.list` 中仍保留为注释,避免同时被系统和 Node 执行。其他可移植规则仍使用系统 crond。
## 夏令时差异
采用 node-cron 的当地时间语义:春季跳时不存在的时间跳过,秋季重复小时不再执行第二次。旧 cron-parser 对部分春季时间会顺延,并可能重复秋季高频任务。使用有夏令时的时区时需要考虑该行为变化;Asia/Shanghai 不受影响。回归测试固定 America/New_York 的 2026 年转换日期。
## 验证
`npm test` 和 `npm run build:back`。
- `test/fixtures/legacy-cron.json` 是迁移前使用 node-schedule 2.1.1 / cron-parser 4.9.0 保存的快照:269 条语法判定,18 组日历序列,每组 10 次执行。
- 真正的 node-cron 实例配合虚拟时钟验证日期/星期 OR、交集去重、闰年、L/#、8 秒迟到补执行、取消释放、回调拒绝和夏令时。
- 注册构造失败回滚、无效订阅隔离、空表达式立即运行,以及现有调度恢复/503/子进程测试覆盖接入流程。
- 隔离的正式 Debian 镜像使用实际编译产物验证 10,000 条有效任务和 2 条错误任务的恢复、300 个同时到期的 Python 脚本、并发上限 16 及 HTTP 健康检查。具体结果见本次测试报告;这是生产镜像隔离验证,不是线上长期运行证明。
### 本次验收结果(2026-09-27)
完整回归 260 通过、3 项原有跳过,后端编译成功。Node 20 生产运行时的 21 项兼容测试通过。隔离镜像中 300/300 脚本完成,无重复或遗漏,最大并发 16,270 个成功及 30 个预设失败均落库;完成耗时 52.36 秒。恢复后健康检查 1,870/1,870 成功,任务列表 38/38 成功。重启恢复阶段另有 50 次健康检查失败。
初版重复解析曾导致恢复超过 5 秒 RPC 时限,加入有上限的语法缓存后重跑通过。本次 10,000 个任务共享两个表达式,不能视为一万种复杂表达式的容量保证;实际脚本耗时也不代表纯调度性能。
新增宏在 UTC 和 Asia/Shanghai 下验证连续两次实际触发及不提前触发;原有语法快照仅对这三个新增宏允许预期变化。
### 复审后验收
完整回归 267 通过、3 项原有跳过;Node 20 的 10 项修复回归及 Alpine 镜像的 24 项兼容/宏测试通过。
额外使用 9,700 个不同表达式和 300 个同时到期任务验证恢复(另有 2 个无效任务),不再局限于共享两个表达式。300/300 脚本完成,无重复或遗漏,最大并发 16,270 个退出码 0、30 个预设退出码 3 均落库。恢复后健康检查 1,650/1,650 成功、任务列表 33/33 成功;启动恢复阶段另有 260 次健康检查失败,整批脚本耗时 60.27 秒。较大快照恢复仍需等待,不承诺启动阶段始终可用或任意规模都能在时限内恢复。
+54
View File
@@ -0,0 +1,54 @@
# Task startup and notification loading
`task` performs shell configuration, status reporting and dependency-path discovery
before invoking the selected runtime. These costs are separate from cron callback
latency and time spent waiting for an execution slot.
## Dependency-path cache
Node scheduling usually starts in `/ql`, while Alpine crond starts in `/root`.
The pnpm discovery key includes the working directory, npm/pnpm configuration,
relevant environment and executable metadata. Each key now has its own cached
record, so alternating scheduler environments do not evict each other.
Records expire after 60 seconds. Refreshes share a persistent per-user lock;
records older than an hour are removed on refresh. Warm lookups do not scan the
cache directory. The lock is never removed while another process might use it.
The opt-out `QL_NODE_PATH_CACHE=0`, failed-discovery fallback, and package resolution
order remain available. This caches the dependency directory, not its contents.
## Language startup paths
| Invocation | Environment and before hooks | Automatic notification loading |
| --- | --- | --- |
| `task script.py` / `.pyc` | Python preload imports generated environment, runs the shell/command before hooks, imports `task_before.py`, then applies account selection | `QLAPI.notify(...)` imports `__ql_notify__` on first call |
| `task script.js` / `.mjs` / `.ts` | Node preload imports generated environment, runs shell/command before hooks, requires `task_before.js`, then applies account selection | `QLAPI.notify(...)` requires `__ql_notify__.js` on first call |
| `task script.sh` | Shell sources the generated environment and runs shell before hooks | No automatic Python/Node notification module |
| Explicit interpreter or other command, e.g. `task python3 script.py` | Existing generic-command path uses shell environment/before hooks; it does not automatically install the Python/Node preload | No automatic notification module unless configured by the caller |
TypeScript uses `ts-node-transpile-only`; MJS uses the same Node preload plus the
ESM loader. There are no separate built-in Ruby, Go, Java or PHP preload modules.
All these task paths still pass through shared dependency-path setup: non-JS
scripts and their hooks may themselves call Node or use the QLAPI bridge.
The before hooks are deliberately still synchronous: they can export environment
variables or perform work required by the user's script. Python's preload still
starts a child shell/Python to capture that environment; Node's preload starts a
child shell/Node. Node also still loads its gRPC client eagerly. These are remaining
startup costs, not work performed by the cron library, and are not removed by this
change.
## Lazy notifications
Scripts keep calling `QLAPI.notify` with the same arguments. When no notification
is sent, the notification provider and its channel dependencies are never imported
by the preload. First use loads the provider; subsequent uses reuse the standard
Python/Node module cache. JS forwards the receiver and return value, including
promises, and Python forwards positional and keyword arguments. Import/send errors
surface at the notification call rather than preventing unrelated scripts from
obtaining QLAPI during startup. A failed import can be retried after its cause is
resolved.
Provider initialization, including provider configuration reads, now happens at
first use. A script's explicit `import __ql_notify__`, `require(...)`, or independent
notification helper still loads that module immediately.
+1 -3
View File
@@ -90,7 +90,6 @@
"compression": "^1.7.4",
"cors": "^2.8.5",
"cron-parser": "^5.4.0",
"cron-parser-v4": "npm:cron-parser@4.9.0",
"cross-spawn": "^7.0.6",
"dayjs": "^1.11.13",
"dotenv": "^16.4.6",
@@ -108,7 +107,7 @@
"keyv": "^5.2.3",
"lodash": "^4.17.21",
"multer": "2.1.1",
"node-schedule": "^2.1.0",
"node-cron": "4.6.0",
"nodemailer": "^8.0.1",
"p-queue-cjs": "7.3.4",
"proper-lockfile": "^4.1.2",
@@ -145,7 +144,6 @@
"@types/lodash": "^4.14.185",
"@types/multer": "^1.4.7",
"@types/node": "^17.0.21",
"@types/node-schedule": "^1.3.2",
"@types/nodemailer": "^6.4.4",
"@types/proper-lockfile": "^4.1.4",
"@types/ps-tree": "^1.1.6",
+8 -44
View File
@@ -42,9 +42,6 @@ dependencies:
cron-parser:
specifier: ^5.4.0
version: 5.4.0
cron-parser-v4:
specifier: npm:cron-parser@4.9.0
version: /cron-parser@4.9.0
cross-spawn:
specifier: ^7.0.6
version: 7.0.6
@@ -96,9 +93,9 @@ dependencies:
multer:
specifier: 2.1.1
version: 2.1.1
node-schedule:
specifier: ^2.1.0
version: 2.1.1
node-cron:
specifier: 4.6.0
version: 4.6.0
nodemailer:
specifier: ^8.0.1
version: 8.0.1
@@ -203,9 +200,6 @@ devDependencies:
'@types/node':
specifier: ^17.0.21
version: 17.0.45
'@types/node-schedule':
specifier: ^1.3.2
version: 1.3.2
'@types/nodemailer':
specifier: ^6.4.4
version: 6.4.17
@@ -4516,12 +4510,6 @@ packages:
'@types/express': 4.17.21
dev: true
/@types/node-schedule@1.3.2:
resolution: {integrity: sha512-Y0CqdAr+lCpArT8CJJjJq4U2v8Bb5e7ru2nV/NhDdaptCMCRdOL3Y7tAhen39HluQMaIKWvPbDuiFBUQpg7Srw==}
dependencies:
'@types/node': 17.0.45
dev: true
/@types/node@17.0.45:
resolution: {integrity: sha512-w+tIMs3rq2afQdsPJlODhoUEKzFP1ayaoyl1CcnwtIlsVe7K7bA1NGm4s3PraqTLlXnbIN84zuBlxBWo1u9BLw==}
@@ -7046,13 +7034,6 @@ packages:
resolution: {integrity: sha512-VQ2MBenTq1fWZUH9DJNGti7kKv6EeAuYr3cLwxUWhIu1baTaXh4Ib5W2CqHVqib4/MqbYGJqiL3Zb8GJZr3l4g==}
dev: true
/cron-parser@4.9.0:
resolution: {integrity: sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==}
engines: {node: '>=12.0.0'}
dependencies:
luxon: 3.5.0
dev: false
/cron-parser@5.4.0:
resolution: {integrity: sha512-HxYB8vTvnQFx4dLsZpGRa0uHp6X3qIzS3ZJgJ9v6l/5TJMgeWQbLkR5yiJ5hOxGbc9+jCADDnydIe15ReLZnJA==}
engines: {node: '>=18'}
@@ -10445,10 +10426,6 @@ packages:
triple-beam: 1.4.1
dev: false
/long-timeout@0.1.1:
resolution: {integrity: sha512-BFRuQUqc7x2NWxfJBCyUrN8iYUYznzL9JROmRz1gZ6KlOIgmoD+njPVbb+VNn2nGMKggMsK79iUNErillsrx7w==}
dev: false
/long@5.2.3:
resolution: {integrity: sha512-lcHwpNoggQTObv5apGNCTdJrO69eHOZMi4BNC+rTLER8iHAqGrUVeLh/irVIM7zTw2bOXA8T6uNPeujwOLg/2Q==}
dev: false
@@ -10482,11 +10459,6 @@ packages:
yallist: 4.0.0
dev: true
/luxon@3.5.0:
resolution: {integrity: sha512-rh+Zjr6DNfUYR3bPwJEnuwDdqMbxZW7LOQfUN4B54+Cl+0o5zaU9RJ6bcidfDtC1cWCZXQ+nvX8bf6bAji37QQ==}
engines: {node: '>=12'}
dev: false
/luxon@3.7.2:
resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==}
engines: {node: '>=12'}
@@ -10849,6 +10821,11 @@ packages:
resolution: {integrity: sha512-73sE9+3UaLYYFmDsFZnqCInzPyh3MqIwZO9cw58yIqAZhONrrabrYyYe3TuIqtIiOuTXVhsGau8hcrhhwSsDIQ==}
dev: false
/node-cron@4.6.0:
resolution: {integrity: sha512-Si/bzYiKRHOB8/a99T2+SDGN582ONDMSTlJr5oCkT6GtnqPjZ2s10eoQRYkW9ZHwjVxONL+W8Fb+qR0AHMQsdg==}
engines: {node: '>=20'}
dev: false
/node-domexception@1.0.0:
resolution: {integrity: sha512-/jKZoMpw0F8GRwl4/eLROPA3cfcXtLApP0QzLmUT/HuPCZWyB7IY9ZrMeKw2O/nFIqPQB3PVM9aYm0F312AXDQ==}
engines: {node: '>=10.5.0'}
@@ -10963,15 +10940,6 @@ packages:
resolution: {integrity: sha512-d9VeXT4SJ7ZeOqGX6R5EM022wpL+eWPooLI+5UpWn2jCT1aosUQEhQP214x33Wkwx3JQMvIm+tIoVOdodFS40g==}
dev: true
/node-schedule@2.1.1:
resolution: {integrity: sha512-OXdegQq03OmXEjt2hZP33W2YPs/E5BcFQks46+G2gAxs4gHOIVD1u7EqlYLYSKsaIpyKCK9Gbk0ta1/gjRSMRQ==}
engines: {node: '>=6'}
dependencies:
cron-parser: 4.9.0
long-timeout: 0.1.1
sorted-array-functions: 1.3.0
dev: false
/nodemailer@8.0.1:
resolution: {integrity: sha512-5kcldIXmaEjZcHR6F28IKGSgpmZHaF1IXLWFTG+Xh3S+Cce4MiakLtWY+PlBU69fLbRa8HlaGIrC/QolUpHkhg==}
engines: {node: '>=6.0.0'}
@@ -14180,10 +14148,6 @@ packages:
sort-object-keys: 1.1.3
dev: true
/sorted-array-functions@1.3.0:
resolution: {integrity: sha512-2sqgzeFlid6N4Z2fUQ1cvFmTOLRi/sEDzSQ0OKYchqgoPmQBVyM3959qYx3fpS6Esef80KjmpgPeEr028dP3OA==}
dev: false
/source-map-js@1.2.1:
resolution: {integrity: sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==}
engines: {node: '>=0.10.0'}
+10 -2
View File
@@ -65,7 +65,9 @@ ql_read_node_path_cache() {
# the caller's descriptors. A crashed refresher releases the kernel lock.
ql_refresh_node_global_path() (
local cache="$1" key="$2" now result previous_umask
local lock="${cache}.lock"
# Keep one persistent lock per owner. Removing per-key lock files could
# split concurrent waiters across different inodes during cache cleanup.
local lock="${dir_tmp}/pnpm-root-${EUID}.lock"
previous_umask=$(umask)
umask 077
if type -P flock &>/dev/null && mkdir -p -- "$dir_tmp" 2>/dev/null; then
@@ -93,6 +95,10 @@ ql_refresh_node_global_path() (
# Private lock creation must not change pnpm's inherited creation mask.
umask "$previous_umask"
# Warm lookups never scan the directory. Retire old environment records on
# refresh, while preserving the shared lock and unrelated cache files.
find "$dir_tmp" -maxdepth 1 -type f -user "$EUID" \
-name "pnpm-root-${EUID}-*.cache" -mmin +60 -delete 2>/dev/null || true
now=${EPOCHSECONDS:-$(date +%s)}
result=$(pnpm root -g 9>&- 2>/dev/null) || return $?
# Never cache failed, empty, multiline or non-absolute answers.
@@ -119,8 +125,10 @@ ql_get_node_global_path() {
fi
local key cache
cache="${dir_tmp}/pnpm-root-${EUID}.cache"
key=$(ql_node_path_cache_key) || { pnpm root -g 2>/dev/null; return $?; }
# Node and crond may have different working directories/configuration.
# Preserve both discoveries instead of evicting each other on every run.
cache="${dir_tmp}/pnpm-root-${EUID}-${key// /-}.cache"
if ql_read_node_path_cache "$cache" "$key"; then
return 0
fi
+6 -2
View File
@@ -159,9 +159,13 @@ try {
run();
const { sendNotify } = require('./__ql_notify__.js');
global.QLAPI = {
notify: sendNotify,
notify(...args) {
// Keep notification dependencies out of ordinary script startup.
// require caches the module after the first notification.
const { sendNotify } = require('./__ql_notify__.js');
return sendNotify.apply(this, args);
},
...client,
};
} catch (error) {
+4 -2
View File
@@ -129,10 +129,12 @@ try:
run()
from __ql_notify__ import send
class BaseApi(Client):
def notify(self, *args, **kwargs):
# Most scripts never send a notification. Python caches this
# module after the first call, including its channel dependencies.
from __ql_notify__ import send
return send(*args, **kwargs)
QLAPI = BaseApi()
+51
View File
@@ -0,0 +1,51 @@
const test = require('node:test');
const assert = require('node:assert/strict');
const { createCronJob } = require('../../back/shared/cronScheduler');
const { isValidCronSchedule } = require('../../back/shared/cronSchedule');
// node-cron caches timezone formatters; model each deployment in a fresh process.
if (!process.env.QL_MACRO_TEST_TZ) {
for (const tz of ['UTC', 'Asia/Shanghai']) {
test(`macro execution in ${tz}`, () => {
const result = require('node:child_process').spawnSync(process.execPath,
['-r', 'ts-node/register/transpile-only', '--test', __filename],
{ env: { ...process.env, TZ: tz, QL_MACRO_TEST_TZ: tz }, encoding: 'utf8' });
assert.equal(result.status, 0, result.stdout + result.stderr);
});
}
}
for (const tz of process.env.QL_MACRO_TEST_TZ ? [process.env.QL_MACRO_TEST_TZ] : []) {
for (const [macro, start, next] of [
['@annually', [2026, 11, 31, 23, 59, 59], [[2027, 0, 1, 0, 0, 0], [2028, 0, 1, 0, 0, 0]]],
['@midnight', [2026, 8, 27, 23, 59, 59], [[2026, 8, 28, 0, 0, 0], [2026, 8, 29, 0, 0, 0]]],
['@minutely', [2026, 8, 27, 12, 34, 59], [[2026, 8, 27, 12, 35, 0], [2026, 8, 27, 12, 36, 0]]],
]) {
test(`${macro} fires at the expected local times in ${tz}`, async (t) => {
const previousTZ = process.env.TZ;
process.env.TZ = tz;
t.after(() => { if (previousTZ === undefined) delete process.env.TZ; else process.env.TZ = previousTZ; });
t.mock.timers.enable({ apis: ['Date', 'setTimeout', 'setInterval'], now: new Date(...start) });
const expected = next.map((parts) => new Date(...parts).getTime());
const actual = [];
assert.equal(isValidCronSchedule(` ${macro} `), true);
const job = createCronJob(macro, (date) => actual.push(date.getTime()), {
name: macro, logger: { warn() {}, error() {} },
});
t.after(() => job.cancel());
for (const time of expected) {
t.mock.timers.tick(time - Date.now() - 1);
await new Promise(setImmediate);
assert.equal(actual.includes(time), false, 'must not fire early');
t.mock.timers.tick(1);
await new Promise(setImmediate);
}
assert.deepEqual(actual, expected);
});
}
}
test('new aliases do not enable unknown macros or raw parser extensions', () => {
for (const schedule of ['@annuallyx', '@midnightx', '@minutelyx', '@secondly', '@reboot', 'H * * * *', '0 /5 * * * *']) {
assert.equal(isValidCronSchedule(schedule), false, schedule);
}
});
+63 -35
View File
@@ -1,41 +1,69 @@
const test = require('node:test');
const assert = require('node:assert/strict');
const nodeSchedule = require('node-schedule');
const { isValidCronSchedule } = require('../../back/shared/cronSchedule');
const { createCronJob } = require('../../back/shared/cronScheduler');
const legacy = require('../fixtures/legacy-cron.json');
const logger = { warn() {}, error() {} };
const flush = () => new Promise(setImmediate);
test('validation agrees with the actual scheduler across legacy cron syntax', () => {
const candidates = new Set([
'*', '0', '?', '0 0', '0 0 *', '0 0 * *',
'@yearly', '@annually', '@monthly', '@weekly', '@daily', '@midnight',
'@hourly', '@secondly', '@minutely', '@weekdays', '@weekends', '@reboot',
'0 0 1 1 * 2027', '0 0 0 L * *', '0 0 0 * * 5L',
'0 0 0 * * MON#2', '0 0 12 LW * *', '0 0 12 15W * *',
'0 0 0 ? * MON', '0 0/30 * * * ?', '0 /5 * * * *',
]);
const fields = [
['*', '?', '/5', '*/5', '0/5', '0', '01', '1-5', '1,3', '5-1', 'H', 'H/5', 'H(0-10)', 'L'],
['*', '?', '/5', '*/5', '0/5', '0', '01', '1-5', '1,3', '5-1', 'H', 'H/5', 'H(0-10)', 'L'],
['*', '?', '/2', '*/2', '0/2', '0', '01', '1-5', '1,3', '23-2', 'H', 'L'],
['*', '?', '/2', '*/2', '1/2', '1', '01', '1-5', '1,3', 'L', 'L-1', 'LW', '15W', 'H'],
['*', '?', '/2', '*/2', '1/2', '1', '01', '1-5', '1,3', 'JAN', 'jan', 'JAN-MAR', 'DEC-FEB', 'H'],
['*', '?', '/2', '*/2', '0/2', '0', '7', '01', '1-5', '1,3', 'MON', 'mon', 'MON-FRI', 'FRI-MON', '5L', 'L', 'MON#2', '1#5', 'H'],
];
fields.forEach((options, index) => options.forEach((field) => {
const values = ['0', '0', '0', '*', '*', '*'];
values[index] = field;
candidates.add(values.join(' '));
}));
for (const value of [...candidates]) {
if (value.split(' ').length === 6) {
candidates.add(` ${value} `);
candidates.add(value.split(' ').slice(1).join(' '));
}
test('validation preserves legacy syntax except the explicitly added macros', () => {
const addedMacros = new Set(['@annually', '@midnight', '@minutely']);
assert.ok(legacy.validation.length > 250);
for (const { schedule, accepted } of legacy.validation) {
assert.equal(isValidCronSchedule(schedule), accepted || addedMacros.has(schedule), schedule);
}
for (const value of candidates) {
const job = nodeSchedule.scheduleJob(value, () => {});
const accepted = Boolean(job);
job?.cancel();
assert.equal(isValidCronSchedule(value), accepted, value);
}
assert.ok(candidates.size > 250);
});
for (const sample of legacy.times) {
test(`calendar compatibility: ${sample.schedule} from ${sample.currentDate}`, async (t) => {
const previousTZ = process.env.TZ;
process.env.TZ = 'UTC';
t.after(() => { if (previousTZ === undefined) delete process.env.TZ; else process.env.TZ = previousTZ; });
t.mock.timers.enable({ apis: ['Date', 'setTimeout', 'setInterval'], now: new Date(sample.currentDate) });
const actual = [];
const job = createCronJob(sample.schedule, (date) => actual.push(date.toISOString()), { name: 'calendar', logger });
t.after(() => job.cancel());
for (const expected of sample.next) {
t.mock.timers.tick(new Date(expected).getTime() - Date.now());
await flush();
}
assert.deepEqual(actual, sample.next);
});
}
test('late callbacks catch up exactly once and cancellation destroys native tasks', async (t) => {
t.mock.timers.enable({ apis: ['Date', 'setTimeout', 'setInterval'], now: new Date('2026-09-27T00:00:00Z') });
const native = require('node-cron');
const before = native.getTasks().size;
const dates = [], warnings = [];
const job = createCronJob('* * * * * *', (date) => dates.push(date.getTime()), {
name: 'late', logger: { warn: (...args) => warnings.push(args), error() {} },
});
t.after(() => job.cancel());
t.mock.timers.tick(8000);
await flush();
assert.equal(dates.length, 8);
assert.equal(new Set(dates).size, 8);
assert.ok(warnings.length > 0);
job.cancel();
t.mock.timers.tick(5000);
await flush();
assert.equal(dates.length, 8);
assert.equal(native.getTasks().size, before);
});
test('callback rejection is logged and future executions continue', async (t) => {
t.mock.timers.enable({ apis: ['Date', 'setTimeout', 'setInterval'], now: new Date('2026-09-27T00:00:00Z') });
const errors = [];
let count = 0;
const job = createCronJob('* * * * * *', async () => { count++; throw Error('callback failed'); }, {
name: 'failure', logger: { warn() {}, error: (...args) => errors.push(args) },
});
t.after(() => job.cancel());
t.mock.timers.tick(1000);
await flush();
t.mock.timers.tick(1000);
await flush();
assert.equal(count, 2);
assert.equal(errors.length, 2);
});
+9 -1
View File
@@ -25,6 +25,14 @@ test('system and node modes route extended cron once and keep portable rules in
const rows = [
{ id: 'plain', schedule: '*/5 0-23 * * 1,3' },
{ id: 'spaces', schedule: ' 0\t0 * * * ' },
{ id: 'numeric-step', schedule: '0/5 * * * *' },
{ id: 'sunday-seven', schedule: '0 0 * * 7' },
{ id: 'day-step', schedule: '0 0 */2 * *' },
{ id: 'restricted-days', schedule: '0 0 1-31 * 1' },
{ id: 'restricted-week', schedule: '0 0 1 * 0-6' },
{ id: 'annually', schedule: '@annually' },
{ id: 'midnight', schedule: '@midnight' },
{ id: 'minutely', schedule: '@minutely' },
{ id: 'seconds', schedule: '0 0 * * * *' },
{ id: 'question', schedule: '0 0 * * ?' },
{ id: 'last', schedule: '0 0 L * *' },
@@ -43,7 +51,7 @@ test('system and node modes route extended cron once and keep portable rules in
}
await service.setCrontab({ data: rows, total: rows.length });
assert.equal(installs, mode === 'system' ? 1 : 0);
for (const id of ['seconds', 'question', 'last', 'nth', 'named', 'macro', 'short', 'extra', 'once', 'boot']) {
for (const id of ['numeric-step', 'sunday-seven', 'day-step', 'restricted-days', 'restricted-week', 'annually', 'midnight', 'minutely', 'seconds', 'question', 'last', 'nth', 'named', 'macro', 'short', 'extra', 'once', 'boot']) {
const line = file.split('\n').find((line) => line.endsWith(`task ${id}.js`));
assert.ok(line.startsWith('# '), `${mode}: ${id} cannot also run from system crontab`);
}
+8 -2
View File
@@ -4,6 +4,8 @@ const { format } = require('node:util');
const load = require('../helpers/load-security-module.cjs');
const { SchedulerReadiness } = require('../../back/shared/schedulerReadiness');
const { getTasks } = require('node-cron');
const legacySchedules = [
'* * * * *',
'*/5 * * * *',
@@ -54,8 +56,11 @@ test('legacy main and extra schedules restore real jobs and healthy readiness',
assert.equal(stacks.size, crons.length);
for (const jobs of stacks.values()) {
assert.equal(jobs.length, 2);
assert.ok(jobs.every((job) => job.nextInvocation()));
assert.ok(jobs.every((job) => typeof job.cancel === 'function'));
}
const nativeJobs = [...getTasks().values()];
assert.equal(nativeJobs.length, crons.length * 2);
assert.ok(nativeJobs.every((job) => job.getStatus() !== 'stopped'));
const previousJobs = [...stacks.values()].flat();
for (const schedule of ['@bogus', '0 /5 * * * ?', '0 70 * * * ?']) {
for (const invalid of [
@@ -64,7 +69,8 @@ test('legacy main and extra schedules restore real jobs and healthy readiness',
]) {
await assert.rejects(register([invalid]), (error) => error.code === 3);
assert.deepEqual([...stacks.values()].flat(), previousJobs);
assert.ok(previousJobs.every((job) => job.nextInvocation()));
assert.deepEqual([...getTasks().values()], nativeJobs);
assert.ok(nativeJobs.every((job) => job.getStatus() !== 'stopped'));
}
}
});
+65
View File
@@ -0,0 +1,65 @@
const test = require('node:test');
const assert = require('node:assert/strict');
const load = require('../helpers/load-security-module.cjs');
const { createCronJob } = require('../../back/shared/cronScheduler');
const logger = { info() {}, warn() {}, error() {} };
test('native calendar skips nonexistent spring time and duplicate fall hour', async (t) => {
const previous = process.env.TZ;
process.env.TZ = 'America/New_York';
t.after(() => { if (previous === undefined) delete process.env.TZ; else process.env.TZ = previous; });
t.mock.timers.enable({ apis: ['Date', 'setTimeout', 'setInterval'], now: new Date('2026-03-08T05:00:00Z') });
const springRuns = [];
const spring = createCronJob('0 30 2 * * *', (date) => springRuns.push(date.toISOString()), { name: 'spring', logger });
t.after(() => spring.cancel());
t.mock.timers.tick(new Date('2026-03-09T06:30:00Z') - Date.now());
await new Promise(setImmediate);
assert.deepEqual(springRuns, ['2026-03-09T06:30:00.000Z']);
spring.cancel();
t.mock.timers.setTime(new Date('2026-11-01T05:30:00Z').getTime());
const fallRuns = [];
const fall = createCronJob('0 30 1 * * *', (date) => fallRuns.push(date.toISOString()), { name: 'fall', logger });
t.after(() => fall.cancel());
t.mock.timers.tick(new Date('2026-11-02T06:30:00Z') - Date.now());
await new Promise(setImmediate);
assert.deepEqual(fallRuns, ['2026-11-02T06:30:00.000Z']);
});
test('construction failure cleans staged jobs and preserves the running snapshot', async () => {
const cancelled = [];
const stacks = new Map([['old', [{ cancel: () => cancelled.push('old') }]]]);
let count = 0;
const { addCron } = load('back/schedule/addCron.ts', {
'./data': { scheduleStacks: stacks },
'../shared/cronScheduler': { createCronJob: () => {
if (++count === 2) throw Error('construction failed');
return { start() { throw Error('must not start'); }, cancel: () => cancelled.push('staged') };
} },
'../shared/runCron': {}, '../loaders/logger': logger, '../shared/i18n': { tf: require('node:util').format },
});
await assert.rejects(new Promise((resolve, reject) => addCron({ request: {
replace: true, crons: [{ id: 'new', schedule: '* * * * *', extra_schedules: [{ schedule: '0 * * * *' }] }],
} }, (error) => error ? reject(error) : resolve())), /construction failed/);
assert.deepEqual([...stacks.keys()], ['old']);
assert.deepEqual(cancelled, ['staged']);
});
test('invalid subscription schedules are isolated and immediate runs without a schedule survive', async () => {
const warnings = [], runs = [];
const Service = load('back/services/schedule.ts', {
typedi: { Service: () => (x) => x, Inject: () => () => {} },
'../shared/pLimit': {},
}).default;
const service = new Service({ ...logger, warn: (...args) => warnings.push(args) });
service.runTask = async (...args) => { runs.push(args); };
const item = { id: 123, name: 'subscription', command: 'true', runOrigin: 'subscription' };
await service.createCronTask({ ...item, schedule: 'not a cron' });
assert.equal(warnings.length, 1);
assert.equal(service.scheduleStacks.size, 0);
await service.createCronTask(item, {}, true);
assert.equal(runs.length, 1);
await service.createCronTask({ ...item, schedule: '* * * * *' });
assert.equal(service.scheduleStacks.size, 1);
await service.cancelCronTask(item);
assert.equal(service.scheduleStacks.size, 0);
});
+29
View File
@@ -65,6 +65,35 @@ test('config, environment, cwd, and executable changes invalidate discovery', (t
f.run();
assert.equal(f.count(), before + 1);
});
test('alternating Node/crond working directories retain independent cached roots', (t) => {
const f = fixture(t);
fs.mkdirSync(path.join(f.root, 'child'));
fs.writeFileSync(path.join(f.root, 'child', '.npmrc'), 'global-dir=/child\n');
for (let i = 0; i < 4; i++) {
assert.equal(f.run({ ANSWER: '/parent/modules' }).stdout.trim(), '/parent/modules');
assert.equal(
f.run({ ANSWER: '/child/modules' }, 'cd child; ' + f.command).stdout.trim(),
'/child/modules',
);
}
assert.equal(f.count(), 2, 'each working directory should discover only once');
assert.equal(fs.readdirSync(f.env.dir_tmp).filter((x) => x.endsWith('.cache')).length, 2);
});
test('refresh retires old per-environment records without removing locks or unrelated files', (t) => {
const f = fixture(t);
fs.mkdirSync(f.env.dir_tmp);
const prefix = `pnpm-root-${process.getuid()}`;
const names = [`${prefix}-1-2.cache`, `${prefix}.lock`, 'unrelated.cache'];
const old = new Date(Date.now() - 2 * 60 * 60 * 1000);
for (const name of names) {
const file = path.join(f.env.dir_tmp, name);
fs.writeFileSync(file, '');
fs.utimesSync(file, old, old);
}
assert.equal(f.run().status, 0);
assert.equal(fs.existsSync(path.join(f.env.dir_tmp, names[0])), false);
for (const name of names.slice(1)) assert.ok(fs.existsSync(path.join(f.env.dir_tmp, name)));
});
test('expired or malformed records refresh and failed lookups are not cached', (t) => {
const f = fixture(t);
f.run();
+6 -2
View File
@@ -51,7 +51,11 @@ function fixture(t) {
fs.existsSync(env.CALLS)
? fs.readFileSync(env.CALLS, 'utf8').trim().split('\n').length
: 0;
const cache = path.join(env.dir_tmp, `pnpm-root-${process.getuid()}.cache`);
const key = run({}, '. "$HELPER"; ql_node_path_cache_key').stdout.trim();
const cache = path.join(
env.dir_tmp,
`pnpm-root-${process.getuid()}-${key.replace(' ', '-')}.cache`,
);
return {
root,
bin,
@@ -61,7 +65,7 @@ function fixture(t) {
asyncRun,
count,
cache,
lock: cache + '.lock',
lock: path.join(env.dir_tmp, `pnpm-root-${process.getuid()}.lock`),
};
}
function check(r, output = '/test/global/node_modules') {
+168
View File
@@ -0,0 +1,168 @@
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 { spawnSync } = require('node:child_process');
function fixture(t, language) {
const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ql-lazy-notify-'));
t.after(() => fs.rmSync(root, { recursive: true, force: true }));
const extension = language === 'python' ? 'py' : 'js';
fs.copyFileSync(`shell/preload/sitecustomize.${extension}`, path.join(root, `sitecustomize.${extension}`));
fs.writeFileSync(path.join(root, `env.${extension}`), '');
fs.writeFileSync(path.join(root, 'before.sh'), 'export FROM_SHELL=before\n');
fs.writeFileSync(path.join(root, 'before.js'), 'process.env.FROM_LANGUAGE = "javascript";\n');
fs.writeFileSync(path.join(root, 'task_before.py'), 'import os\nos.environ["FROM_LANGUAGE"] = "python"\n');
fs.writeFileSync(path.join(root, 'client.js'), 'module.exports = { getEnvs: () => "client-ready" };\n');
fs.copyFileSync('shell/preload/client.py', path.join(root, 'client.py'));
fs.copyFileSync('shell/preload/esm-loader.mjs', path.join(root, 'esm-loader.mjs'));
const env = {
...process.env,
PYTHONPATH: root,
PREV_PYTHONPATH: '',
NODE_OPTIONS: '',
PREV_NODE_OPTIONS: '',
QL_NODE_GLOBAL_PATH: '',
file_task_before: path.join(root, 'before.sh'),
file_task_before_js: path.join(root, 'before.js'),
dir_scripts: root,
task_before: 'export FROM_COMMAND=command',
envParam: 'ACCOUNTS',
numParam: '2-3',
ACCOUNTS: 'first&second&third',
};
const write = (name, text) => fs.writeFileSync(path.join(root, name), text);
const run = (body, extensionOverride = extension) => {
const entry = path.join(root, `entry.${extensionOverride}`);
write(path.basename(entry), body);
const python = language === 'python';
const result = spawnSync(python ? 'python3' : process.execPath,
python ? [entry] : ['--require', path.join(root, 'sitecustomize.js'), entry],
{ cwd: root, env, encoding: 'utf8', timeout: 15000 });
assert.equal(result.status, 0, result.stderr + result.stdout);
return result.stdout;
};
return { root, env, write, run };
}
for (const extension of ['js', 'mjs']) {
test(`${extension} startup preserves hooks, environment selection and client without importing notifications`, (t) => {
const f = fixture(t, 'javascript');
f.write('__ql_notify__.js', 'throw Error("notification dependencies must not load at startup");');
f.run(`
const assert = ${extension === 'mjs' ? '(await import("node:assert/strict")).default' : 'require("node:assert/strict")'};
assert.equal(process.env.FROM_SHELL, 'before');
assert.equal(process.env.FROM_LANGUAGE, 'javascript');
assert.equal(process.env.FROM_COMMAND, 'command');
assert.equal(process.env.ACCOUNTS, 'second&third');
assert.equal(QLAPI.getEnvs(), 'client-ready');
assert.equal(typeof QLAPI.notify, 'function');
`, extension);
});
}
test('JavaScript first notification imports once and preserves arguments, this, promises and errors', (t) => {
const f = fixture(t, 'javascript');
f.write('__ql_notify__.js', `
global.notificationLoads = (global.notificationLoads || 0) + 1;
exports.sendNotify = function (...args) {
if (args[0] === 'fail') throw new Error('send failed');
global.notificationResult = Promise.resolve({ args, receiver: this });
return global.notificationResult;
};
`);
f.run(`
const assert = require('node:assert/strict');
assert.equal(global.notificationLoads, undefined);
(async () => {
const options = { channel: 'fixture' };
const promise = QLAPI.notify('title', 'content', options);
assert.equal(promise, global.notificationResult);
const result = await promise;
assert.deepEqual(result.args, ['title', 'content', options]);
assert.equal(result.receiver, QLAPI);
await QLAPI.notify('second');
assert.equal(global.notificationLoads, 1);
assert.throws(() => QLAPI.notify('fail'), /send failed/);
})().catch(error => { console.error(error); process.exitCode = 1; });
`);
});
test('JavaScript reports import failures on notify and permits a later retry', (t) => {
const f = fixture(t, 'javascript');
f.write('__ql_notify__.js', `
if (!process.env.NOTIFY_READY) throw Error('notification dependency unavailable');
exports.sendNotify = (...args) => args;
`);
f.run(`
const assert = require('node:assert/strict');
assert.equal(QLAPI.getEnvs(), 'client-ready');
assert.throws(() => QLAPI.notify('first'), /notification dependency unavailable/);
process.env.NOTIFY_READY = '1';
assert.deepEqual(QLAPI.notify('retry'), ['retry']);
`);
});
test('Python startup preserves hooks, environment selection and client without importing notifications', (t) => {
const f = fixture(t, 'python');
f.write('__ql_notify__.py', 'raise RuntimeError("notification dependencies must not load at startup")\n');
f.run(`
import os, sys
assert os.environ['FROM_SHELL'] == 'before'
assert os.environ['FROM_LANGUAGE'] == 'python'
assert os.environ['FROM_COMMAND'] == 'command'
assert os.environ['ACCOUNTS'] == 'second&third'
assert '__ql_notify__' not in sys.modules
assert callable(QLAPI.notify)
assert callable(QLAPI.getEnvs)
`);
});
test('Python first notification imports once and preserves args, kwargs, return values and errors', (t) => {
const f = fixture(t, 'python');
f.write('__ql_notify__.py', `
import builtins
builtins.notification_loads = getattr(builtins, 'notification_loads', 0) + 1
def send(*args, **kwargs):
if args[0] == 'fail':
raise ValueError('send failed')
return args, kwargs
`);
f.run(`
import builtins, sys
assert '__ql_notify__' not in sys.modules
assert QLAPI.notify('title', 'content', channel='fixture') == (('title', 'content'), {'channel': 'fixture'})
assert QLAPI.notify('second') == (('second',), {})
assert builtins.notification_loads == 1
try:
QLAPI.notify('fail')
except ValueError as error:
assert str(error) == 'send failed'
else:
raise AssertionError('notification exception was swallowed')
`);
});
test('Python reports import failures on notify and permits a later retry', (t) => {
const f = fixture(t, 'python');
f.write('__ql_notify__.py', `
import os
if not os.environ.get('NOTIFY_READY'):
raise RuntimeError('notification dependency unavailable')
def send(*args, **kwargs):
return args
`);
f.run(`
import os, sys
assert '__ql_notify__' not in sys.modules
try:
QLAPI.notify('first')
except RuntimeError as error:
assert str(error) == 'notification dependency unavailable'
else:
raise AssertionError('import exception was swallowed')
os.environ['NOTIFY_READY'] = '1'
assert QLAPI.notify('retry') == ('retry',)
`);
});
@@ -0,0 +1,40 @@
const test = require('node:test');
const assert = require('node:assert/strict');
const grpc = require('@grpc/grpc-js');
const load = require('../helpers/load-security-module.cjs');
const { SchedulerReadiness } = require('../../back/shared/schedulerReadiness');
for (const [name, rows, delay, succeeds] of [
['large distinct snapshot', Array.from({length:10000}, (_, i) => ({id:String(i), schedule:`${i%60} ${Math.floor(i/60)%60} ${Math.floor(i/3600)} * * *`})), 10000, true],
['extra schedules', [{id:'extra', schedule:'* * * * *', extra_schedules:Array.from({length:2000},()=>({schedule:'0 * * * *'}))}], 10000, true],
['unresponsive small write', [{id:'small', schedule:'* * * * *'}], 30000, false],
['unresponsive large write remains bounded', Array.from({length:30000},()=>({schedule:'* * * * *'})), 180000, false],
]) {
test(`recovery deadline: ${name}`, async (t) => {
t.mock.timers.enable({apis:['Date','setTimeout'], now:0});
let calls=0, deadline;
const fake={waitForReady:(_,cb)=>cb(), addCron:(_req,_meta,options,cb)=>{
calls++; deadline=options.deadline;
const budget=deadline-Date.now();
setTimeout(()=>cb(delay<=budget ? null : Object.assign(Error('deadline'),{code:grpc.status.DEADLINE_EXCEEDED}), {}), Math.min(delay,budget));
}};
const client=load('back/schedule/client.ts', {
'../protos/cron':{CronClient:class{constructor(){return fake}}},
'../config':{grpcPort:5500},
'../config/grpcCerts':{getGrpcCerts:()=>({caCert:'ca',clientKey:'key',clientCert:'cert'})},
'@grpc/grpc-js':{...grpc,credentials:{createSsl:()=>({})}},
}).default;
const readiness=new SchedulerReadiness(async()=>{}, 60000);
t.after(()=>clearTimeout(readiness.retry));
client.readiness=readiness;
readiness.configure(()=>client.addCron(rows,true));
const recovering=readiness.recover();
await new Promise(setImmediate);
assert.equal(calls,1);
assert.ok(deadline<=120000);
t.mock.timers.tick(Math.min(delay,deadline));
assert.equal(await recovering,succeeds);
assert.equal(calls,1,'uncertain writes are not replayed inline');
assert.equal(await readiness.check(),succeeds);
});
}
+4 -2
View File
@@ -41,8 +41,10 @@ test('recovery reconciles a surviving scheduler after missed deletes and disable
);
const { addCron } = load('back/schedule/addCron.ts', {
'./data': { scheduleStacks: stacks },
'node-schedule': {
scheduleJob: (id) => ({ cancel: () => cancelled.push(id) }),
'../shared/cronScheduler': {
createCronJob: (_schedule, _callback, options) => ({
start() {}, cancel: () => cancelled.push(options.name.split(':')[0]),
}),
},
'../shared/runCron': {},
'../loaders/logger': { info() {}, warn() {} },
+1371
View File
File diff suppressed because it is too large Load Diff