jason810496 commented on code in PR #76103:
URL: https://github.com/apache/airflow/pull/76103#discussion_r4547868480
##########
ts-sdk/src/coordinator/runtime.ts:
##########
@@ -380,9 +405,11 @@ async function handleTask(
try {
switch (task.kind) {
case "operator":
- return await runOperator(
- task.operator,
- buildOperatorContext({ details, dag: task.dag, ctx, client, logs,
fail }),
+ return await runTimed(() =>
+ runOperator(
+ task.operator,
+ buildOperatorContext({ details, dag: task.dag, ctx, client, logs,
fail }),
+ ),
);
Review Comment:
A timed-out operator goes through `fail` and logs "Task failed" without the
stack, while a handler logs "Task timed out". Could the catch below match
`runHandler`?
```ts
} catch (err) {
if (err instanceof TaskTimeoutError) {
logs.error("Task timed out", { task_id: ctx.taskId, error:
err.message, stack: err.stack ?? null });
return buildFailureResponse(details, err.message);
}
return fail((err as Error).message ?? String(err));
}
```
##########
ts-sdk/src/coordinator/execution-timeout.ts:
##########
@@ -0,0 +1,115 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// A task's `execution_timeout`, enforced inside the runtime so the task can
stop
+// at the timeout and fail with it. The supervisor keeps its own deadline for a
+// process that cannot stop itself, such as one stuck in a synchronous loop.
+
+import { TaskTimeoutError } from "../sdk/task.js";
+
+// setTimeout fires at once for a longer delay, so the supervisor alone
enforces
+// a timeout past this (about 24.8 days).
+const MAX_TIMER_MS = 2 ** 31 - 1;
+
+// What the timer resolves with: no value a task returns can equal it.
+const TIMED_OUT = Symbol("timed out");
+
+/**
+ * The task's timeout in seconds, from its run context, or `null` when it has
+ * none: unset and `0` both mean no timeout.
+ */
+export function executionTimeoutOf(tiContext: {
+ execution_timeout?: number | null;
+}): number | null {
+ const seconds = tiContext.execution_timeout;
+ return seconds === undefined || seconds === null || seconds === 0 ? null :
seconds;
+}
+
+/** One task's execution timeout: the signal it aborts, and the wait that
enforces it. */
+export interface ExecutionTimeout {
+ /** The timeout in seconds. A negative one has already passed. */
+ readonly seconds: number;
+ /** Aborted with a {@link TaskTimeoutError} when the timeout passes. */
+ readonly signal: AbortSignal;
+ /**
+ * Wait for the task, which is already running; the timeout counts from here.
+ *
+ * At the timeout {@link signal} is aborted, so a task passing it to its work
+ * stops there. The task then has the cleanup time to finish; what it returns
+ * or throws by then is ignored, so a task that times out never succeeds. A
+ * task that ignores the signal is not waited for longer: the timeout is
+ * reported, and the supervisor ends the process if it lingers.
+ */
+ wait<T>(running: Promise<T>): Promise<T>;
+}
+
+/** How long {@link ExecutionTimeout.wait} waits for a task after its timeout.
*/
+export interface ExecutionTimeoutOptions {
+ /** How long the task has to finish after the timeout. */
+ readonly cleanupMs: number;
+ /**
+ * Ends that wait early: the runtime's signal, which the supervisor's SIGTERM
+ * aborts when its own deadline comes first.
+ */
+ readonly stopWaiting: AbortSignal;
+}
+
+export function createExecutionTimeout(
+ seconds: number,
+ { cleanupMs, stopWaiting }: ExecutionTimeoutOptions,
+): ExecutionTimeout {
+ const controller = new AbortController();
+ return {
+ seconds,
+ signal: controller.signal,
+ async wait<T>(running: Promise<T>): Promise<T> {
+ const timeoutMs = seconds * 1000;
+ if (timeoutMs > MAX_TIMER_MS) return running;
+
+ let timer: ReturnType<typeof setTimeout> | undefined;
+ const timedOut = new Promise<typeof TIMED_OUT>((resolve) => {
+ timer = setTimeout(() => resolve(TIMED_OUT), timeoutMs);
+ });
Review Comment:
Why do we return "running" promise if timeoutMs is larger than MAX_TIMER_MS?
Past about 24.8 days the runtime enforces nothing, so a handler that returns
after the supervisor's SIGTERM still succeeds. Re-arming the timer in steps
keeps the timeout enforced for any length:
```suggestion
let timer: ReturnType<typeof setTimeout> | undefined;
const timedOut = new Promise<typeof TIMED_OUT>((resolve) => {
const arm = (leftMs: number) => {
timer =
leftMs > MAX_TIMER_MS
? setTimeout(() => arm(leftMs - MAX_TIMER_MS), MAX_TIMER_MS)
: setTimeout(() => resolve(TIMED_OUT), leftMs);
};
arm(timeoutMs);
});
```
##########
ts-sdk/src/coordinator/runtime.ts:
##########
@@ -452,6 +484,27 @@ async function runHandler(
}
}
+/** Starts a task's own work and waits for it under its execution_timeout, if
it has one. */
+type RunTimed = <T>(start: () => Promise<T>) => Promise<T>;
+
+/**
+ * Tell the supervisor the task's timeout now, so its own deadline counts from
the
+ * same moment as the runtime's. A supervisor that rejects the message keeps
the
+ * deadline it set when the runtime started.
+ */
+async function sendTimeoutToSupervisor(
+ comm: CommChannel,
+ logs: LogChannel,
+ timeoutSeconds: number,
+): Promise<void> {
+ const err = parseFrameError(
+ await comm.request({ type: "SetExecutionTimeout", timeout_seconds:
timeoutSeconds }),
+ );
+ if (err) {
+ logs.warning("Supervisor did not take SetExecutionTimeout", { error:
err.code });
+ }
Review Comment:
`comm.request` throws on a closed channel or after 30 s with no reply, so a
send failure here fails the task before it starts. Should that warn as well,
like the error-frame case?
```suggestion
try {
const err = parseFrameError(
await comm.request({ type: "SetExecutionTimeout", timeout_seconds:
timeoutSeconds }),
);
if (err) {
logs.warning("Supervisor did not take SetExecutionTimeout", { error:
err.code });
}
} catch (err) {
logs.warning("Could not send SetExecutionTimeout", {
error: (err as Error).message ?? String(err),
});
}
```
##########
ts-sdk/src/coordinator/execution-timeout.ts:
##########
@@ -0,0 +1,115 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// A task's `execution_timeout`, enforced inside the runtime so the task can
stop
+// at the timeout and fail with it. The supervisor keeps its own deadline for a
+// process that cannot stop itself, such as one stuck in a synchronous loop.
+
+import { TaskTimeoutError } from "../sdk/task.js";
+
+// setTimeout fires at once for a longer delay, so the supervisor alone
enforces
+// a timeout past this (about 24.8 days).
+const MAX_TIMER_MS = 2 ** 31 - 1;
+
+// What the timer resolves with: no value a task returns can equal it.
+const TIMED_OUT = Symbol("timed out");
+
+/**
+ * The task's timeout in seconds, from its run context, or `null` when it has
+ * none: unset and `0` both mean no timeout.
+ */
+export function executionTimeoutOf(tiContext: {
+ execution_timeout?: number | null;
+}): number | null {
+ const seconds = tiContext.execution_timeout;
+ return seconds === undefined || seconds === null || seconds === 0 ? null :
seconds;
+}
+
+/** One task's execution timeout: the signal it aborts, and the wait that
enforces it. */
+export interface ExecutionTimeout {
+ /** The timeout in seconds. A negative one has already passed. */
+ readonly seconds: number;
+ /** Aborted with a {@link TaskTimeoutError} when the timeout passes. */
+ readonly signal: AbortSignal;
+ /**
+ * Wait for the task, which is already running; the timeout counts from here.
+ *
+ * At the timeout {@link signal} is aborted, so a task passing it to its work
+ * stops there. The task then has the cleanup time to finish; what it returns
+ * or throws by then is ignored, so a task that times out never succeeds. A
+ * task that ignores the signal is not waited for longer: the timeout is
+ * reported, and the supervisor ends the process if it lingers.
+ */
+ wait<T>(running: Promise<T>): Promise<T>;
+}
+
+/** How long {@link ExecutionTimeout.wait} waits for a task after its timeout.
*/
+export interface ExecutionTimeoutOptions {
+ /** How long the task has to finish after the timeout. */
+ readonly cleanupMs: number;
+ /**
+ * Ends that wait early: the runtime's signal, which the supervisor's SIGTERM
+ * aborts when its own deadline comes first.
+ */
+ readonly stopWaiting: AbortSignal;
+}
+
+export function createExecutionTimeout(
+ seconds: number,
+ { cleanupMs, stopWaiting }: ExecutionTimeoutOptions,
+): ExecutionTimeout {
+ const controller = new AbortController();
+ return {
+ seconds,
+ signal: controller.signal,
+ async wait<T>(running: Promise<T>): Promise<T> {
+ const timeoutMs = seconds * 1000;
+ if (timeoutMs > MAX_TIMER_MS) return running;
+
+ let timer: ReturnType<typeof setTimeout> | undefined;
+ const timedOut = new Promise<typeof TIMED_OUT>((resolve) => {
+ timer = setTimeout(() => resolve(TIMED_OUT), timeoutMs);
+ });
+ let first: T | typeof TIMED_OUT;
+ try {
+ first = await Promise.race([running, timedOut]);
+ } finally {
+ clearTimeout(timer);
+ }
+ if (first !== TIMED_OUT) return first;
+
+ const error = new TaskTimeoutError(seconds);
+ controller.abort(error);
+ // Until the task finishes, its cleanup time runs out, or the
supervisor's
+ // SIGTERM arrives, whichever comes first.
+ await new Promise<void>((resolve) => {
+ const done = () => {
+ clearTimeout(cleanupTimer);
+ stopWaiting.removeEventListener("abort", done);
+ resolve();
+ };
+ const cleanupTimer = setTimeout(done, cleanupMs);
+ stopWaiting.addEventListener("abort", done);
+ if (stopWaiting.aborted) done();
+ running.then(done, done);
+ });
+ throw error;
Review Comment:
A handler that ignores the signal keeps running after this throws. The
runtime then reports the failure and closes comm, so its later `getClient()`
calls fail on the closed channel, and the process stays up until the supervisor
kills it after `task_success_overtime`. Should the runtime exit once it has
reported a timeout?
##########
ts-sdk/src/coordinator/execution-timeout.ts:
##########
@@ -0,0 +1,115 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// A task's `execution_timeout`, enforced inside the runtime so the task can
stop
+// at the timeout and fail with it. The supervisor keeps its own deadline for a
+// process that cannot stop itself, such as one stuck in a synchronous loop.
+
+import { TaskTimeoutError } from "../sdk/task.js";
+
+// setTimeout fires at once for a longer delay, so the supervisor alone
enforces
+// a timeout past this (about 24.8 days).
Review Comment:
With the timer re-armed, this comment changes too:
```suggestion
// setTimeout fires at once for a delay past this (about 24.8 days), so a
longer
// timeout re-arms the timer in steps of at most this.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
