运行后台进程

启动一个在请求结束后仍继续运行的进程，例如构建、测试套件或 Agent 任务。exec() 返回的进程对象属于启动它的请求，因此后续请求通过 Sandbox 中的文件管理该进程。进程会将其进程 ID、输出和退出码写入这些文件。

对于会一直运行到容器停止的进程（例如开发服务器或隧道），请参阅在后台运行服务器。

前置条件

一个带有 Durable Object 的 Worker，该 Durable Object 使用 Durable Object 调度策略启动容器。要创建一个，请参阅运行 Linux 命令。

在后台运行进程

添加用于运行进程并报告其状态的 shell 脚本：

src/index.jsjs

const ROOT = "/var/lib/processes";
const INACTIVITY_TIMEOUT_MS = 10 * 60 * 1000;

// Runs the command in its own process group. Records its process ID when
// it starts and its exit code when it ends.
const RUN = `dir=$1; shift
setsid sh -c 'echo "$$ $(cat /proc/sys/kernel/random/boot_id)" >"$0/pid"; exec "$@"' \\
    "$dir" "$@" >"$dir/stdout.log" 2>"$dir/stderr.log"
echo "$?" >"$dir/exit-code.tmp" && mv "$dir/exit-code.tmp" "$dir/exit-code"`;

// Defines current(), which reads the process ID in a directory into $pid
// when the process started in this instance.
const CURRENT = `current() {
    read -r pid boot 2>/dev/null <"$1/pid" &&
        [ "$boot" = "$(cat /proc/sys/kernel/random/boot_id)" ]
}`;

// Prints the state of the process that owns the directory.
const STATUS = `${CURRENT}
dir=$1
if [ ! -d "$dir" ]; then echo missing
elif [ -e "$dir/exit-code" ]; then echo "exited $(cat "$dir/exit-code")"
elif [ ! -e "$dir/pid" ]; then echo starting
elif current "$dir" && kill -0 "$pid" 2>/dev/null; then echo "running $pid"
elif [ -e "$dir/exit-code" ]; then echo "exited $(cat "$dir/exit-code")"
else echo lost
fi`;

async function run(container, argv) {
	const process = await container.exec(argv);
	const output = await process.output();

	return {
		exitCode: output.exitCode,
		stdout: new TextDecoder().decode(output.stdout),
	};
}

src/index.tsts

const ROOT = "/var/lib/processes";
const INACTIVITY_TIMEOUT_MS = 10 * 60 * 1000;

// Runs the command in its own process group. Records its process ID when
// it starts and its exit code when it ends.
const RUN = `dir=$1; shift
setsid sh -c 'echo "$$ $(cat /proc/sys/kernel/random/boot_id)" >"$0/pid"; exec "$@"' \\
    "$dir" "$@" >"$dir/stdout.log" 2>"$dir/stderr.log"
echo "$?" >"$dir/exit-code.tmp" && mv "$dir/exit-code.tmp" "$dir/exit-code"`;

// Defines current(), which reads the process ID in a directory into $pid
// when the process started in this instance.
const CURRENT = `current() {
    read -r pid boot 2>/dev/null <"$1/pid" &&
        [ "$boot" = "$(cat /proc/sys/kernel/random/boot_id)" ]
}`;

// Prints the state of the process that owns the directory.
const STATUS = `${CURRENT}
dir=$1
if [ ! -d "$dir" ]; then echo missing
elif [ -e "$dir/exit-code" ]; then echo "exited $(cat "$dir/exit-code")"
elif [ ! -e "$dir/pid" ]; then echo starting
elif current "$dir" && kill -0 "$pid" 2>/dev/null; then echo "running $pid"
elif [ -e "$dir/exit-code" ]; then echo "exited $(cat "$dir/exit-code")"
else echo lost
fi`;

type Status =
    | { state: "starting" | "lost" }
    | { state: "running"; pid: number }
    | { state: "exited"; exitCode: number };

async function run(container: Container, argv: string[]) {
    const process = await container.exec(argv);
    const output = await process.output();

    return {
        exitCode: output.exitCode,
        stdout: new TextDecoder().decode(output.stdout),
    };
}

setsid 会在新的进程组中启动命令。停止该进程组也会停止命令启动的进程，例如 npm run dev 背后的 node 进程，除非某个进程转移到自己的进程组。脚本会将退出代码写入临时文件，然后重命名该文件，确保读取者不会看到不完整的文件。不要改用 & 启动命令：Shell 会忽略后台作业的 SIGINT。

每个进程在实例磁盘上的 /var/lib/processes 下各自保留目录。日志会持续增长，直到进程结束。磁盘写满后，进程的写入会失败，因此请删除旧的进程目录，或将需要保留的输出发送到已挂载的存储桶。

为 Durable Object 添加构造函数。当重启后的 Durable Object 发现容器仍在运行时，它会再次设置不活动超时：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    constructor(ctx: DurableObjectState, env: Env) {
        super(ctx, env);
        const container = ctx.container;
        if (container?.running) {
            void ctx.blockConcurrencyWhile(() =>
                container.setInactivityTimeout(INACTIVITY_TIMEOUT_MS),
            );
        }
    }
}

然后添加启动进程的方法：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async startProcess(id: string, argv: string[]): Promise<boolean> {
        const container = this.ctx.container;

        if (!container) {
            throw new Error("No container is configured");
        }

        if (!container.running) {
            container.start({
                image: "cloudflare/debian-trixie",
                entrypoint: ["sleep", "infinity"],
                enableInternet: false,
            });
            await container.setInactivityTimeout(INACTIVITY_TIMEOUT_MS);
        }

        // mkdir fails when another process already uses the ID.
        const dir = `${ROOT}/${id}`;
        await run(container, ["mkdir", "-p", ROOT]);
        const created = await run(container, ["mkdir", dir]);

        if (created.exitCode !== 0) {
            return false;
        }

        // Ignoring the output lets the process outlive this request.
        await container.exec(["sh", "-c", RUN, "sh", dir, ...argv], {
            stdout: "ignore",
            stderr: "ignore",
        });

        // exec() returns before the process records its ID. Wait for it, so
        // a request that follows the output right away finds it running.
        for (let i = 0; i < 50; i++) {
            if ((await this.status(id))?.state !== "starting") break;
            await scheduler.wait(100);
        }

        await this.ctx.storage.put(`process:${id}`, argv);
        await this.ctx.storage.setAlarm(Date.now() + 60_000);
        return true;
    }
}

添加报告进程状态的方法：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async status(id: string): Promise<Status | undefined> {
        const container = this.ctx.container;

        if (!container?.running) {
            return undefined;
        }

        const { stdout } = await run(container, [
            "sh",
            "-c",
            STATUS,
            "sh",
            `${ROOT}/${id}`,
        ]);
        const [state, value] = stdout.trim().split(" ");

        if (state === "running") {
            return { state, pid: Number(value) };
        }

        if (state === "exited") {
            return { state, exitCode: Number(value) };
        }

        if (state === "starting" || state === "lost") {
            return { state };
        }

        return undefined;
    }
}

kill -0 会检查进程是否存在，而不向其发送信号。当进程结束却未记录退出码时，状态为 lost，例如有人向记录退出码的脚本发送了 SIGKILL。

进程 ID 文件还会记录 boot ID，该 ID 在每个实例中都会变化。snapshot 会还原该文件，而新实例会重用相同的进程 ID，因此还原后的进程 ID 可能属于无关进程。current() 会忽略来自其他实例的进程 ID，因此拍摄 snapshot 时正在运行的进程会变成 lost。

添加一个方法，返回进程截至目前的输出：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async logs(id: string, stream: "stdout" | "stderr"): Promise<string | undefined> {
        if (!(await this.status(id))) {
            return undefined;
        }

        const path = `${ROOT}/${id}/${stream}.log`;
        return (await run(this.ctx.container!, ["cat", path])).stdout;
    }
}

如需在进程运行时流式查看输出，或等待它打印一行内容，请参阅跟踪后台进程。

添加停止进程的方法：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async stop(id: string): Promise<boolean> {
        const status = await this.status(id);

        if (status?.state !== "running") {
            return false;
        }

        // `kill` is a shell built-in. A negative ID signals the process group.
        const { exitCode } = await run(this.ctx.container!, [
            "sh",
            "-c",
            'kill -s TERM -- "$1"',
            "sh",
            `-${status.pid}`,
        ]);
        return exitCode === 0;
    }
}

被信号停止的进程会以 128 加上信号编号退出，例如 SIGTERM 对应 143。

添加 alarm 处理程序，在任何进程运行时保持 Sandbox 运行：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async alarm(): Promise<void> {
        let running = false;

        for (const [key, argv] of await this.ctx.storage.list<string[]>({
            prefix: "process:",
        })) {
            const id = key.slice("process:".length);
            const status = await this.status(id);

            if (status?.state === "running" || status?.state === "starting") {
                running = true;
            } else {
                console.log({ event: "process.ended", id, argv, status });
                await this.ctx.storage.delete(key);
            }
        }

        if (running) {
            await this.ctx.storage.setAlarm(Date.now() + 60_000);
        }
    }
}

正在运行的进程不会让容器保持运行。alarm 每分钟检查每个进程，每次 alarm 都会保持容器运行。在两次 alarm 之间重启的 Durable Object 会在构造函数中再次设置 10 分钟不活动超时。alarm 发现最后一个进程已结束后 10 分钟，容器会停止。更多信息，请参阅 Sandbox 生命周期。

Durable Object 只有一个 alarm。如果你的类已有 alarm() 处理程序，请将进程检查合并进去，并将 alarm 设为任一检查所需的最早时间。更多信息，请参阅 Alarms。

进程结束时，alarm 会在一分钟内向 Workers Logs 写入 process.ended 日志。将 console.log() 调用替换为你的应用接下来要做的事，例如发送通知。

在 Worker 中添加调用这些方法的路由：

src/index.jsjs

const command = [
	"sh",
	"-c",
	"echo ready; for i in $(seq 1 300); do echo tick $i; sleep 1; done",
];

export default {
	async fetch(request, env) {
		const url = new URL(request.url);
		const match = /^\/processes\/([a-z0-9-]{1,63})(\/logs)?$/.exec(
			url.pathname,
		);

		if (!match) {
			return new Response("Not found", { status: 404 });
		}

		const [, id, logs] = match;
		const sandbox = env.MY_CONTAINER.getByName("sandbox");

		if (logs) {
			const stream =
				url.searchParams.get("stream") === "stderr" ? "stderr" : "stdout";
			const output = await sandbox.logs(id, stream);
			return output === undefined
				? new Response("Not found", { status: 404 })
				: new Response(output);
		}

		if (request.method === "POST") {
			return (await sandbox.startProcess(id, command))
				? new Response(null, { status: 201 })
				: new Response("The process ID is in use", { status: 409 });
		}

		if (request.method === "DELETE") {
			return (await sandbox.stop(id))
				? new Response(null, { status: 202 })
				: new Response("The process is not running", { status: 409 });
		}

		const status = await sandbox.status(id);
		return status
			? Response.json(status)
			: new Response("Not found", { status: 404 });
	},
};

src/index.tsts

const command = [
    "sh",
    "-c",
    "echo ready; for i in $(seq 1 300); do echo tick $i; sleep 1; done",
];

export default {
    async fetch(request: Request, env: Env): Promise<Response> {
        const url = new URL(request.url);
        const match = /^\/processes\/([a-z0-9-]{1,63})(\/logs)?$/.exec(
            url.pathname,
        );

        if (!match) {
            return new Response("Not found", { status: 404 });
        }

        const [, id, logs] = match;
        const sandbox = env.MY_CONTAINER.getByName("sandbox");

        if (logs) {
            const stream =
                url.searchParams.get("stream") === "stderr" ? "stderr" : "stdout";
            const output = await sandbox.logs(id, stream);
            return output === undefined
                ? new Response("Not found", { status: 404 })
                : new Response(output);
        }

        if (request.method === "POST") {
            return (await sandbox.startProcess(id, command))
                ? new Response(null, { status: 201 })
                : new Response("The process ID is in use", { status: 409 });
        }

        if (request.method === "DELETE") {
            return (await sandbox.stop(id))
                ? new Response(null, { status: 202 })
                : new Response("The process is not running", { status: 409 });
        }

        const status = await sandbox.status(id);
        return status
            ? Response.json(status)
            : new Response("Not found", { status: 404 });
    },
} satisfies ExportedHandler<Env>;

此示例命令会打印 ready，然后每秒打印一行，持续五分钟。将其替换为你自己的命令。该路由只接受路径中安全的进程 ID，因为每个 ID 都会成为目录名。

部署 Worker：
npx wrangler deploy
yarn wrangler deploy
pnpm wrangler deploy
启动名为 ticker 的进程，然后检查它。将示例主机名替换为 Wrangler 打印的 workers.dev URL：
```
curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/processes/ticker --request POST
curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/processes/ticker
```
第一个请求返回 201，使用相同 ID 的第二次 POST 返回 409。状态包含进程 ID：
```
{ "state": "running", "pid": 354 }
```
读取截至目前的输出：
```
curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/processes/ticker/logs
```
请求会打印截至目前的行，然后结束：
```
ready
tick 1
tick 2
tick 3
```
在查询中添加 stream=stderr 以读取标准错误。

停止进程，然后再次检查：

curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/processes/ticker --request DELETE
curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/processes/ticker

DELETE 请求会返回 202。状态中会显示退出代码：

{ "state": "exited", "exitCode": 143 }

运行后台进程

前置条件

在后台运行进程

相关资源