@chamvi/频道-cron

Chamvi cron频道:每分一分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每分每一次,每分每分每分每分每分每分每一次,每分每分每分每分每分每一次,每分每分每分每分每分每分每分每一次,每分每分每分每分每分每分一次,一次,每分每分每分每分每分每分每分每分每分每分每.
每个复制品都登记在同一频道。 商店的锁每分一分,
同一套装为库伯内特斯克龙乔布发了一发.

目录

• 安装
• 使用量
• 滴答滴答的动作
• 重叠政策
• 时区
• 关闭
• 一发工作
• 选项
• API 密码
• 许可证

安装

npm install @chamvi/channel-cron @chamvi/core --save

仅限无害环境管理,节点.js >=26.0.0。 。 。 . @chamvi/core 是同龄人依赖; hubwire (车内)和 cron-parser 与软件包一起安装.

使用量

cron.on() 订阅 cron 频道的代理; cronChannel() 引擎启动的频道是 :

import type { LanguageModel } from 'ai';
import {
    Engine,
    createMemoryStore,
    defineModelAgent,
} from '@chamvi/core';
import { cron, cronChannel } from '@chamvi/channel-cron';

export async function startScheduling(model: LanguageModel) {
    const sweep = defineModelAgent({
        name: 'sweep',
        model,
        instructions: 'Close stale issues older than 30 days.',
        // every 30 minutes, Berlin time; skipped while the previous run is active
        channels: [cron.on('*/30 * * * *', { timezone: 'Europe/Berlin' })],
    });

    // createMemoryStore is per process; with several replicas, pass one shared store
    // (e.g. createDatabaseStore) so its lock fires each tick once across them
    const engine = new Engine({
        store: createMemoryStore(),
        agents: [sweep],
        // every replica registers the same channel; it reads agents and store from the engine
        channels: [cronChannel()],
    });

    // starts the cron channel after recovering runs
    await engine.start();

    // shutdown: drain() stops the channel first, then lets the runs in flight finish
    // await engine.drain();
    return { engine };
}

cron.on(expression, options?) 验证表达式( cron-parser) 和时区并丢弃 InvalidAgentDefinitionError 对于一个它拒绝. engine.start() 恢复运行后启动频道: 订阅调度器无法使用的代码( 表达式 croner 拒绝) 的代理失败 engine.start() 与 InvalidAgentDefinitionError 命名代理和订阅,代理文件夹的地址是 onError 留下那个特工 其他频道的订阅被忽略. 该频道跟随在运行时间注册,替换并移除的代理.

第二个cron信道(例如每个区域一个)需要自己的名称 on() 只让订阅点火:

const eu = cronChannel({ name: 'cron-eu' });
defineModelAgent({ name: 'digest', channels: [eu.on('0 8 * * *', { timezone: 'Europe/Berlin' })] /* ... */ });
new Engine({ channels: [cronChannel(), eu] /* ... */ });

滴答滴答的动作

• 每个订阅一个部分. 每一个都变成一个 集线器计划组件 命名 cron:<agent>:<subscriptionId>。 。 。 .
• 每滴一个复制品. 在滴答运行之前,复制品会获得商店锁 cron:<agent>:<subscriptionId>:<scheduledAt ISO> (通过) ILockRepository用于 lockTtlMs (默认10分). 密钥永远不会被放出:一个晚点放出同一滴的复制品无法再次取胜.
• 重复发送. 胜者通过频道发送(信封) channel =频道名称, trigger = 有效载荷的订阅id) options.payload ?? { scheduledAt: '<ISO>' } 和独有性键 <subscriptionId>@<scheduledAt ISO>,所以即使同一勾选的第二次射击也返回了现有的运行。 订户自己买的 idempotencyKey() 被忽略为克龙.
• 从来没有开始. 启动或重新启动复制点火无事可做; 第一次发送是下一个预定的勾选 .
• 无相接取. 掉入重启的滴答,或者其计时器起火的滴答 lockTtlMs 或更晚(一个已暂停的流程),不会在以后发射.
• 报告失败,而不是抛出. 锁定存储失败, 重叠检查或发送到达 onError(error, { agent, subscription });下一勾选正常运行.

重叠政策

cron.on(expression, { overlap }) 设置策略; 频道在发送前执行。 "Active"的意思是同一订阅的一跑在 queued, (中文(简体) ), running, (中文(简体) ), waiting 或者说 interrupted。 。 。 .

A 类 concurrencyKey() 在订阅时胜过默认。 默认密钥是 queue-one 等待 : 引擎每次启动一次运行, 一次启动密钥, 所以队列运行在活动者和解时启动 。 这也使得比赛变得无害:在之前的滴滴调度还在飞行中时进行检查的勾子可能会发出第二次跑道,然后等待而不是同时跑道.

时区

cron.on(expression, { timezone: 'Europe/Berlin' }) 取IANA区;没有IANA区,则应用过程的局部区. 在 DST 更改时 :

• 当地时间发生两次(倒回)火灾一次,第一次发生时.
• 当地时间春前变化跳过火灾一次,一小时后当地时间(30 2 * * * 3点半起火).

关闭

engine.drain() (和 engine.stop())首先停止航道(没有新的滴入;飞行完成时滴入),然后排出跑道:

await engine.drain(); // stops the cron channel, waits for running runs, then interrupts the rest

一发工作

runJob() 开出一次Cron订阅 等待它的运行, 一个Kubernetes CronJob 而不是一个长寿的复制品:

const { exitCode } = await runJob({ engine, agent: sweep });
process.exitCode = exitCode;

它在当前时间发送勾选( 重叠策略适用) , 启动引擎 start({ channels: false }) (所以一个工作从不开始 cron 调度或其他渠道) 并运行一个批准勾选(当勾选被跳过时) , 等待到运行 completed, (中文(简体) ), failed, (中文(简体) ), cancelled 或者说 waiting,或 queued 背着同样的订阅 等待批准,然后打电话 engine.drain()。引擎必须持有订阅的 cron 通道(它拒绝代理).

暂停审批的跑步不持有吊舱:工作返回,批准得到回答(或时断出),被答复的跑步由长生复制或下个工作吊舱执行,即使自己的滴滴被跳过也会启动引擎. 这个吊舱还恢复了一次运行 吊舱被杀死 并否认批准过他们的超时。 洞穴:

• engine.start() 追回和执行 每个 已排队运行 每个 代理在共享商店, 不仅仅是克龙运行.
• drain() 中断任何仍在运行的 drainTimeoutMs。 。 。 . runJob 等待它发送的运行, 所以剩余运行完成 由长寿的复制品 或后期的工作舱, 在该舱内 drainTimeoutMs。 。 。 .
• 大小 drainTimeoutMs 更喜欢长期部署 cronChannel() 当运行是漫长的.

选项

cronChannel(options)数字 :

cron.on(expression, options)数字 :

runJob(options)数字 :

API 密码

类型 : CronChannel, (中文(简体) ), CronChannelOptions, (中文(简体) ), CronChannelErrorContext, (中文(简体) ), CronSubscriber, (中文(简体) ), CronSubscriptionOptions, (中文(简体) ), OverlapPolicy, (中文(简体) ), RunJobOptions, (中文(简体) ), RunJobResult, (中文(简体) ), CronOptions, (中文(简体) ), CronTickContext, (中文(简体) ), CronTickResult, (中文(简体) ), CronTickDispatched, (中文(简体) ), CronTickSkipped。 。 。 .

许可证

原载 Apache 许可证 2.0。 。 。 .