监视器( W)@fungi.computer/watchdog(中文(简体) )

守望犬是一个小型耐用的工作引擎. 你交给它,它储存在你的 SQLite 数据库, 每个 tick() 拥有一份工作,运行您的执行人 记录一个最终结果 如果过程在中途死亡, 下一勾 问你的活体探测器发生了什么 并恢复了工作.

npm install @fungi.computer/watchdog

然后运行下面的快速开始: 一个短文件, 内存 node:sqlite 在节点上 24,没有其他设置.

监视者从不看工作 有效载荷、队列名称和车道不透明 数值。 你的密码决定了工作的意义 以及如何运作. 守望犬拥有生命周期:排队,运行,完成,安顿. 索赔,索赔 符号和在状态之间移动行的 SQL 保持对软件包的隐私.

需要工作时用它来活命 一次跑一次 以诚实的结果, 不写自己的国家机器。 它运行在任何 同步 SQLite 拥有者, 如 Cloudflare 持久对象的存储或 node:sqlite,并在共享的 D1 或 PostgreSQL 数据库上.

要求和切入点

监视者只是无害环境管理。 SQLite 切入点运行于您有同步的任何地方 SQLite 控件, 如 Cloudflare 持久对象存储或 node:sqlite 内 节点24。 它的运行时间依赖 包括凯恩, effect, (中文(简体) ), pg, (中文(简体) ), @types/pg 财务报告和审定财务报表 @cloudflare/workers-types,该 npm 为您安装.

条目:

• @fungi.computer/watchdog承诺API, Watchdog.make,与公众 工作类型和计划.
• @fungi.computer/watchdog/effect: 与效果值相同的运行时间,通过 makeRuntime。 。 。 .
• @fungi.computer/watchdog/node-sqlite数字 : fromNodeSqlite(database) (单位:千美元) 调用者拥有的节点SQLite数据库. 它回来了 NodeSqliteOwner; 其编号 NodeSqliteCursor 提供 toArray() 财务报告和审定财务报表 one(),准确返回一个 为零行或多行解码并投出.
• @fungi.computer/watchdog/d1: 在迁移的 Cloudflare D1 上运行时间 数据库.
• @fungi.computer/watchdog/postgres: 在迁移的 PostgreSQL 上运行时间 数据库.

快开始

这工作要经过记忆 node:sqlite 数据库。 这个 owner 对象是 SQLite 切入点的整个主机合同:运行语句, 并运行同步交易.

import { DatabaseSync } from "node:sqlite";
import { fromNodeSqlite } from "@fungi.computer/watchdog/node-sqlite";
import { Watchdog } from "@fungi.computer/watchdog";

const database = new DatabaseSync(":memory:");
const owner = fromNodeSqlite(database);

const watchdog = await Watchdog.make({
  owner,
  execute: async (job, signal) => {
    if (signal.aborted) return { kind: "cancelled" };
    console.log(`sending ${job.jobId}`, job.payload);
    return { kind: "completed" };
  },
  // This process owns every claim, so a running job is alive while it runs.
  isAlive: async () => true,
  // Arm your real timer or alarm here. Ticking by hand is enough for a demo.
  wake: { recompute: async () => {} },
});

await watchdog.ingest({
  jobId: "welcome-email-1",
  queue: "email",
  lane: "default",
  priority: 0,
  payload: { to: "[email protected]" },
  state: "queued",
});

const tick = await watchdog.tick();
console.log(tick.status); // "settled"

const read = await watchdog.read("welcome-email-1");
if (read.status === "found" && read.job.state === "settled")
  console.log(read.job.outcome); // "completed"

database.close();

运行 tick() 它又回来了 { status: "no_ready" }获得同样的工作 它又回来了 { status: "duplicate" }。 。 。 .

在持久目标中, ctx.storage 已经是 SQLite 的拥有者 .

核心概念

作业. 工作有个 jobId页:1 queue页:1 lane整数 priority,一个 不透明 JSON payload 和可选 recovery: { maxRecoveries, windowMs? }。 。 。 . ID和队列名称为1至256个字符;道为1至512个字符. 所有者-当地建筑可预留一车道同时交付;否则 车道没有列表政策。 工作经过四个州:

• queued:接受并等待.
• running: 以一分勾举并交给你的执行人.
• completing:决定执行结果;结算和预测 待定.
• settled: 完成,带有 outcome 联 合 国 completed, (中文(简体) ), failed, (中文(简体) ), cancelled, (中文(简体) ), outcome_unknown 或者说 interrupted,可选 failureReason。 。 。 .

无相是一相. ingest(job) 返回时 queued 新的身份证和 duplicate 同样的重播。 使用同一ID和不同的重放 队列、车道、优先级、有效载荷或回收政策失败.

一个滴答,一个工作. tick() 选择一个排队头, 排得最高的 优先,最古老的,跨合格队列。 它声称你的工作,运行你 执行者,并解决它所花费的确切索赔。 一秒钟 tick() 同样的 飞行返回时运行时间 { status: "busy" } 而不是运行 平行进行。 您的执行者返回其中之一 completed, (中文(简体) ), failed (与 可选短数 reason, (中文(简体) ), cancelled 或者说 outcome_unknown如果掷出或正在投出 中断了, 监视者解决工作为 failed 以它自己的理由代码,这样 作为 execution_failed 或者说 execution_interrupted。 。 。 .

主办口岸. 你提供四样东西:

• owner:快速启动时显示的 SQLite 能力.
• execute(job, signal):经营一份声称的工作。 阁下 AbortSignal。 。 。 .
• isAlive(job): 回答运行中的工作执行是否仍然存在.
• wake.recompute(): 在持久状态变化后, 手持下一个计时器或提醒 .

完成回收. 确切的索赔首先将结果记录为: completing单项交易解决了这一决定并援引了所有人 预测。 如果此项交易失败, 则决定仍可持久 。 每个勾选 在完成新工作之前再试一次。 失败表面 立即; 主机拥有有界的后退. 重新开始从不执行 再次完成工作,探索其活性或收取其回收预算。 队列 头部报告为 running 直至达成和解; read(jobId) 报告 completing。 。 。 .

失踪赔偿. 完成后,将检查最老的 运行索赔。 若为 isAlive 返回时 false,处决已经消失。 一个工作与 恢复预算左转 queued。否则它就成为 interrupted如果 isAlive 抛出, 监视者不会猜测。 索赔保持不变 是的,还有一则闲置的滴答报告 { status: "liveness_unknown", jobId, reason }。 。 。 .

可选正数 windowMs 过了那么长的时间就重新设置计数 自从第一个充电回收 在当前窗口。 时钟回滚不会 重设. 没有窗口,计数仍然是这项工作的终身预算.

效应所有者关闭. 效果运行时间 close 停止发送和 中断它自己的执行。 执行者范围结束后, isDrained(job) 必须证明外国清理工作已经完成。 只有那时,守望者才会重新排队 准确的索赔而不收取追偿。 缺勤或假排水的答案 正在运行和失败的索赔 drain_failed; 重新打开视它为 意外失踪 持久的取消胜过移交。 一个 已经确定完成的继续结算。 进一步勾选返回 stopped。 。 。 . 生产呼叫员是Shiitake, 其运行的瞄准镜供应 固定型号/工具 清净相证. 承诺运行时间不会暴露这种生命周期操作.

撤销. cancel({ jobId, cancellationId }) 记录一个持久 取消 。 队列任务结算为 cancelled 在它的下一个滴答。 一个工作 当前运行时间被中断, 并按 cancelled。 。 。 . 重播同样的游戏 cancellationId 返回时 replayed。返回不同的一个 conflict。一旦长期执行决定,取消即为不取消 返回时 { status: "decided", outcome }包括定居后。 过 onlyIfQueued: true 一旦工作开始就拒绝取消.

莱恩拒绝. failLane({ lane, reason }) 已排队的终极和解者, 运行和完成该道的工作 failed,保留每项工作及其 边际所有者理性法. 它每个电话最多处理256个工作,运行工作 首先,然后返回 { settled, more }。仅在 more 真实; 256 是 过渡批次,而不是接受工作的上限。 已经解决的历史和 其他车道则保持不变。 精疲力尽后,再回电话 { settled: 0, more: false } 并且不发射投影仪。 后序为新 工作,而不是永久拒绝标记.

在所有者- 本地的 SQLite 上, 每个更改的工作都引用 project 中标 交易作为其结算。 投影机把整个批次都卷回去 所有所有者投影都写着 。 本地运行的执行器在 长期拒绝;其滴答仍在等待清理,并报告已承诺的失败 而不再次投影。 替换的持久索赔无法覆盖 失败。 拒绝不能抵消已经接受的提供者效力.

共享的 D1/ PostgreSQL 适配器暴露了相同的有界限的结算方法和 在一次原子SQL变异中,栅栏租赁. 他们不提供同步 仅使用 SQL project 调用。 主机仍然拥有权限,必须停止撤销 执行; 监视犬拥有持久的工作结算,而不是App录用.

所有者交易. runtime.transactional 是用于 代码已经在您的 SQLite 交易中 . enqueueAcceptedJob 财务报告和审定财务报表 recordCancellationIntent 使用相同的变异 ingest 财务报告和审定财务报表 cancel这样 您的域写和任务承诺或返回到一起. readQueueHead 财务报告和审定财务报表 readLatestSettledJob 读取交易中的队列状态.

队列投影. 过 project 改为 Watchdog.make 接收 { queue, head, settled? } 在与每笔交易相同的交易中切换 摄取,主张,取消,结算,拒绝道口,恢复和清洗. 使用 它保持每行状态 与守望者。 从一个掷出 project 失败过渡 .

清洗. purge(queue) 删除队列中的每个已队列任务。 运行中, 完成和安置工作的人留下来.

排定的工作和交付努力

队列任务可能携带 notBefore,绝对的毫秒期限。 监视者 保留为 not_before; 未来的队列头在后继者时 独立队列继续. wakeState(now) 返回时 { pending, nextWakeAt } 从同一行,包括共享租赁。 取消绕过到期的未来 时间。 现在主机调度 pending 是真的,否则 nextWakeAt。 。 。 .

执行人可提供代号 retryPolicy 带有边框后退和 或尝试上限或绝对期限。 监视犬增量 attempt_count 并记录处决前的退伍时间。 执行者 接收保留尝试,返回 retry 保留同样的工作 当预算时, 接收到一个已耗尽的上下文而不尝试另一个接收器 结束。 送货失踪案保留了这项指控,不借用 恢复配额。 原生效果执行者 replaySafe 上游为工作 其全部效果为一所有者交易,如延长时间.

现有所有者-当地交货

所有者-当地建筑可选择采用 concurrentLane。仅配置 运行时间类型添加 tickConcurrent():普通勾选排除该道和 第二勾选只选择它。 共享一个商店和交易 进行预测,有单独的忙碌的警卫和积极的处决。 取消 表示拥有此任务的槽; 回收只检查空位的 分区。 两个槽都要求队列头跨道,通过跑步命令 首先,然后优先级降入并排入序列升出. 已取消队列 任务可以不执行而通过繁忙的队列解决.

效果运行时间 readSettled(limit, afterSequence, lane?) 揭露 顺序序列分区 。 一个前缀围栏扣留了以后的定居点 直到所选分区和可选行道中所有较早的工作落地或已落地 已清净. 同时期收款从不消耗一个页段;它们仍可读取 工作证. wakeState(now) 仍然覆盖两个分区. 这些所有者 - 本地 操作不需要存储的策略更改,也不更改 D1 或 PostgreSQL 适配器.

共享 D1 和 PostgreSQL 存储

@fungi.computer/watchdog/d1 出口 createD1Watchdog 财务报告和审定财务报表 watchdogD1Schema。 。 。 . @fungi.computer/watchdog/postgres 出口 createPostgresWatchdog 财务报告和审定财务报表 watchdogPostgresSchema。通过您的迁移应用软件包拥有的计划 在打开运行时间之前使用工具。 打开运行时间从不迁移数据库 。 这个 名称 watchdog_execution_state 制约识别当前生命周期 语法. 过时的语法在被处决前被拒绝. 土著预备 入住仍然是来电者转移交易的一部分;没有执行 修道计划 所有者-当地SQLite在打开时适用同样的拒绝.

这些跑步时间是一样的 ingest, (中文(简体) ), read, (中文(简体) ), readQueueHead, (中文(简体) ), tick, (中文(简体) ), cancel, (中文(简体) ), failLane, (中文(简体) ), purge 财务报告和审定财务报表 wakeState(now) 操作。 他们拿 execute, (中文(简体) ), wake 和可选 now 时钟代替主人和活人 探测器。 守望者从索赔租约中回答活生生的问题.

D1级 enqueue(job) 返回为生产者的本地批次准备的语句. enqueueFromSelect({ text, values }) 接受一个可信任的本地 SQLite ?NNN- 捆绑的SELECT的 job_id, (中文(简体) ), queue_name, (中文(简体) ), lane, (中文(简体) ), priority, (中文(简体) ), payload_json,无效 recovery_max 和无效 recovery_window_ms。 。 。 . 监视犬提供其INSERT和冲突检查. 这让工作依赖于 有条件地插入到同一批次的前面, 没有 JavaScript 读在中间。 邮递SQL enqueue(client, job) 使用呼叫器的检出 交易客户端。 呼叫者拥有BEGIN,COMMIT和ROLLBACK. 同样的 复出无所取. 身份或货物冲突 整个本地人 交易.

两种适配器都使用与SQLite入站点相同的生命周期引擎. 索偿 保持私人状态,有2分钟的租约、10秒钟的续约和有限度的续约 I/O. 执行必须履行中止命令。 拒绝延期和解决 过期或被替换的索赔。 失败的活性探测器从未宣布执行 已经死了。 数据库时间检查阻止飞行中续航复活 过期索赔。 可选 now 时钟取代确定性的数据库时间 测试。 生产主机应当使用默认值. 持续的主机唤醒仍然驱动 滴答和消失后恢复。 供应商的影响依然存在 至少从头开始.

purgeD1Queues 财务报告和审定财务报表 purgePostgresQueues 返回参数化去除sure 对信任的制片人SQL选择的队列的语句。 打电话的人会加入他们 它的原产地批量或交易, 而守望者仍然是工作的唯一所有者 删除 SQL 。 这些帮助删除选中队列中的每次尝试和 没有披露索偿或结算控制. 承运人 编曲 他们带着信封去除.

API 参考

生成的引用在 文档/api覆盖4个 模块 :

• @fungi.computer/watchdog数字 : Watchdog.make, (中文(简体) ), WatchdogRuntime, (中文(简体) ), WatchdogOptions, (中文(简体) ), WatchdogError,以及任务、结果和 SQLite 拥有者类型 他们的效应计划.
• @fungi.computer/watchdog/effect数字 : makeRuntime, (中文(简体) ), WatchdogRuntimeEffect 财务报告和审定财务报表 EffectWatchdogOptions。效果运行时间也有 readSettled(limit, afterSequence?, lane?) 寻找找寻固定工作的人.
• @fungi.computer/watchdog/node-sqlite数字 : fromNodeSqlite(database) (单位:千美元) 调用者拥有的节点SQLite数据库. 它回来了 NodeSqliteOwner; 其编号 NodeSqliteCursor 提供 toArray() 财务报告和审定财务报表 one(),准确返回一个 为零行或多行解码并投出.
• @fungi.computer/watchdog/d1数字 : createD1Watchdog, (中文(简体) ), prepareD1Admission, (中文(简体) ), watchdogD1Schema 财务报告和审定财务报表 purgeD1Queues。 。 。 .
• @fungi.computer/watchdog/postgres数字 : createPostgresWatchdog, (中文(简体) ), watchdogPostgresSchema 财务报告和审定财务报表 purgePostgresQueues。 。 。 .

实例和检查

效果切入点在没有承诺包的情况下运行同一引擎:

import * as Effect from "effect/Effect";
import { makeRuntime } from "@fungi.computer/watchdog/effect";

const runtime = await Effect.runPromise(
  makeRuntime({
    owner, // the same owner as in the quick start
    executor: { execute: () => Effect.succeed({ kind: "completed" as const }) },
    isAlive: () => Effect.succeed(true),
    wake: { recompute: () => Effect.void },
  }),
);
const result = await Effect.runPromise(runtime.tick());

在您自己的交易中, 将它属于的域名排序为 :

owner.transactionSync(() => {
  owner.sql.exec("INSERT INTO orders (id) VALUES (?)", "order-1");
  watchdog.transactional.enqueueAcceptedJob({
    jobId: "charge-order-1",
    queue: "order-1",
    lane: "default",
    priority: 0,
    payload: { orderId: "order-1" },
    state: "queued",
  });
});

套件测试在真实的记忆上运行引擎 node:sqlite 拥有者。 他们 覆盖勾选使用周期, 取消排队和运行中的工作, 消失 恢复、 失败原因、 清理、 队列投影、 交易投影 和共享储存租赁。 从软件包目录 :

pnpm test

限制和错误

每一次失败都是 WatchdogError 关闭 cause数字 :

• store_unavailable:商店抛出,或者一个摄入回放与 存储的工作.
• store_invalid:生命周期计划已过时,或车道-反向输入 其严格的车道/理由解码器失败。 其他商店的断层仍然存在 store_unavailable。 。 。 .
• claim_unavailable: 无法受理或重新提出索赔.
• drain_failed: 执行中断不能证明外国清理 结束了.
• execution_failed: 你的承诺执行者扔了. tick() 不拒绝。 这个 任务结算为 failed 以此代码为它的 failureReason。 。 。 .
• settlement_rejected:索赔在理赔前被替换或过期.
• wake_unavailable数字 : wake.recompute() 抛出.
• liveness_unavailable数字 : isAlive 抛出.

underlying 保留服务器侧诊断的原始抛出值。 别 寄给客户.

限制和保障:

• 一个滴答最多一个工作 打电话 tick() 或从你醒来,到 排出队列.
• 货币是每次运行的时间。 每个拥有者打开一个SQLite运行时间. D1级和 PostgreSQL运行时间使用租赁,这样不止一个工人可以共享一个数据库.
• 交货时间至少是一秒 撞车后又找了份工作,你快点 执行器副作用同位素.
• outcome_unknown 是终极结果。 监视者不会重试的.
• ID和队列名称为1至256个字符. 巷道和 concurrentLane 份额 1个1个至512个字符,横跨就业招生、储存和建筑。 A 类 256个字符需要263个字符 member: 或265个 与 team-key:; 512 留有空间进行配料,而不散开 截断身份. 新鲜的共享数据库计划执行相同的约束. failureReason 1到200个字符,应该是一个简短的代码,永远不要 提供文本.

贡献和许可证

监视者持有麻省理工学院的执照. 见 自由主义。 。 。 .