This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new c6a547ea138 [v3-4-test] TypeScript SDK: take the task state store
retention in seconds (#75445) (#75460)
c6a547ea138 is described below
commit c6a547ea1383f3a8ecf2c0dca0989c3b2aa2895d
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 23 20:04:11 2026 +0800
[v3-4-test] TypeScript SDK: take the task state store retention in seconds
(#75445) (#75460)
* TypeScript SDK: take the task state store retention in seconds
retentionMs was the one duration in the SDK that was not seconds, next to
executionTimeout, retryDelay and dagrunTimeout, which all are. Rename it to
retentionSeconds so every duration carries the same unit.
The HITL example modules now export their Dags, so another bundle can serve
them.
* TypeScript SDK: keep the HITL example Dags unexported
Importing them would run the module-level serve() call, so the importing
bundle's own serve() would then fail.
* TypeScript SDK: name the task state store option retention
Every other public duration is a bare name in seconds, and the Python
argument is retention too.
* TypeScript SDK: report an overflowing retention in seconds
The caller passes seconds, so the error now uses the same unit.
* TypeScript SDK docs: state the seconds rule once in a Durations section
The task state store section links to it instead of carrying the rule.
* TypeScript SDK: name the operator timeout option timeout
Every duration in the SDK is now a bare name in seconds, as defer's
timeout is in Python.
---------
(cherry picked from commit ccbd215ff291801ddbc3599167b4287a5551c161)
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
.../language-sdks/typescript.rst | 25 ++++++++++++-----
ts-sdk/example/src/main.ts | 2 +-
ts-sdk/src/coordinator/client.ts | 32 ++++++++++++----------
ts-sdk/src/coordinator/operator-runner.ts | 4 +--
ts-sdk/src/hitl/operator.ts | 2 +-
ts-sdk/src/sdk/client-types.ts | 10 +++----
ts-sdk/src/sdk/client.ts | 6 ++--
ts-sdk/src/sdk/operator.ts | 8 +++---
ts-sdk/tests/coordinator/client.test.ts | 28 ++++++++++++-------
ts-sdk/tests/coordinator/operator-runner.test.ts | 8 +++---
ts-sdk/tests/public-api.test.ts | 2 +-
11 files changed, 75 insertions(+), 52 deletions(-)
diff --git
a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
index b9937aa8829..7a5699dcb13 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -565,7 +565,7 @@ The ``TaskClient``
* - ``setXCom({ key, value, ... })``
- Publishes an XCom value.
* - ``taskStateStore``
- - This task instance's key/value store, with ``get<T>(key)``, ``set(key,
value, { retentionMs? })``,
+ - This task instance's key/value store, with ``get<T>(key)``, ``set(key,
value, { retention? })``,
``delete(key)`` and ``clear()``. See
:ref:`typescript-sdk/task-state-store`.
.. note::
@@ -595,7 +595,7 @@ for the full concept, including the equivalent Python API.
let jobId = await store.get<string>("job_id");
if (jobId == null) {
jobId = await submitSparkJob();
- await store.set("job_id", jobId, { retentionMs: NEVER_EXPIRE });
+ await store.set("job_id", jobId, { retention: NEVER_EXPIRE });
}
const result = await waitForSparkJob(jobId);
@@ -603,15 +603,17 @@ for the full concept, including the equivalent Python API.
return result;
}
-**Retention.** ``retentionMs`` is milliseconds to retain the key, counted from
the time of the write:
+**Retention.** ``retention`` is how long to keep the key, in seconds (see
:ref:`typescript-sdk/durations`),
+counted from the time of the write:
-* A number retains the key for that many milliseconds from now; ``retentionMs:
0`` expires the key
+* A number retains the key for that many seconds from now; ``retention: 0``
expires the key
immediately.
* ``NEVER_EXPIRE`` (imported from ``apache-airflow-ts-sdk``) stores the key
with no expiry, regardless of
the deployment default.
-* When ``retentionMs`` is omitted the runtime reads
``AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS``, which
- the coordinator sets from ``[state_store] default_retention_days``. A value
of ``0`` there means the key
- never expires. This is not the same as ``retentionMs: 0`` above, which
expires the key immediately.
+* When ``retention`` is omitted the runtime reads
``AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS``,
+ which the coordinator sets from ``[state_store] default_retention_days``. A
value of ``0`` there means the
+ key never expires. This is not the same as ``retention: 0`` above, which
expires the key
+ immediately.
**Keys** must be non-empty strings of at most 512 characters; they may contain
slashes. **Values** must be
JSON-compatible (see :ref:`typescript-sdk/types`), and can never be ``null`` —
``set`` rejects a ``null``
@@ -632,6 +634,15 @@ Anything a task writes to stdout or stderr, such as
``console.log`` or ``console
Airflow task log: stdout at ``INFO`` level, stderr at ``ERROR``. There is no
dedicated structured-logging API
yet.
+.. _typescript-sdk/durations:
+
+Durations
+---------
+
+Every duration in the SDK is a number of seconds, with no unit in its name:
the Dag and task options
+``retryDelay``, ``executionTimeout`` and ``dagrunTimeout``,
``responseTimeout`` on a human-in-the-loop task,
+``pokeInterval`` on ``triggerDagRun``, and ``retention`` in the task state
store.
+
.. _typescript-sdk/types:
XCom type mapping
diff --git a/ts-sdk/example/src/main.ts b/ts-sdk/example/src/main.ts
index 1fd2c38b17a..756ec0fb762 100644
--- a/ts-sdk/example/src/main.ts
+++ b/ts-sdk/example/src/main.ts
@@ -66,7 +66,7 @@ export async function writeAndReadTaskState() {
const store = getClient().taskStateStore;
const { runId, tryNumber } = getContext();
- await store.set(JOB_ID_KEY, runId, { retentionMs: NEVER_EXPIRE });
+ await store.set(JOB_ID_KEY, runId, { retention: NEVER_EXPIRE });
await store.set(STATE_SCRATCH_KEY, { attempt: tryNumber });
const jobId = await store.get<string>(JOB_ID_KEY);
diff --git a/ts-sdk/src/coordinator/client.ts b/ts-sdk/src/coordinator/client.ts
index 025dd29d68c..7584a23e0f9 100644
--- a/ts-sdk/src/coordinator/client.ts
+++ b/ts-sdk/src/coordinator/client.ts
@@ -142,7 +142,8 @@ const TASK_STATE_STORE_ABSENT_POLICY: AbsentRowPolicy = {
// `[state_store] default_retention_days` at launch
// (task-sdk/src/airflow/sdk/coordinators/_subprocess.py).
const RETENTION_DAYS_ENV_VAR = "AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS";
-const MS_PER_DAY = 24 * 60 * 60 * 1000;
+const MS_PER_SECOND = 1000;
+const SECONDS_PER_DAY = 24 * 60 * 60;
function assertKey(key: string): void {
if (typeof key !== "string") {
@@ -157,28 +158,29 @@ function assertKey(key: string): void {
// past 9999, and the supervisor silently drops a frame it cannot decode
// rather than replying with an error, so the task would hang instead of
// failing fast. Route both retention paths through this check.
-function toExpiresAt(ms: number): string {
- const at = new Date(Date.now() + ms);
+function toExpiresAt(seconds: number): string {
+ const at = new Date(Date.now() + seconds * MS_PER_SECOND);
if (Number.isNaN(at.getTime()) || at.getUTCFullYear() > 9999) {
throw new RangeError(
- `retention of ${ms}ms overflows the wire timestamp; use NEVER_EXPIRE
instead`,
+ `a retention that ends ${seconds}s from now overflows the wire
timestamp; ` +
+ "use NEVER_EXPIRE instead",
);
}
return at.toISOString();
}
-/** Resolve `retentionMs` (or the deployment default) to a wire `expires_at`.
*/
-function resolveExpiresAt(retentionMs: number | typeof NEVER_EXPIRE |
undefined): string | null {
- if (retentionMs === undefined) {
+/** Resolve `retention` (or the deployment default) to a wire `expires_at`. */
+function resolveExpiresAt(seconds: number | typeof NEVER_EXPIRE | undefined):
string | null {
+ if (seconds === undefined) {
return resolveDefaultExpiresAt();
}
- if (retentionMs === NEVER_EXPIRE) {
+ if (seconds === NEVER_EXPIRE) {
return null;
}
- if (!Number.isFinite(retentionMs) || retentionMs < 0) {
- throw new RangeError(`retentionMs must be >= 0 or NEVER_EXPIRE, got
${String(retentionMs)}`);
+ if (!Number.isFinite(seconds) || seconds < 0) {
+ throw new RangeError(`retention must be >= 0 or NEVER_EXPIRE, got
${String(seconds)}`);
}
- return toExpiresAt(retentionMs);
+ return toExpiresAt(seconds);
}
// The coordinator always passes the setting, so an absent or malformed value
is
@@ -189,8 +191,8 @@ function resolveDefaultExpiresAt(): string | null {
if (raw === undefined) {
throw new Error(
`${RETENTION_DAYS_ENV_VAR} is not set, so the default retention is
unknown. It carries ` +
- "[state_store] default_retention_days to the runtime; pass retentionMs
or NEVER_EXPIRE to " +
- "set the expiry explicitly.",
+ "[state_store] default_retention_days to the runtime; pass retention
or " +
+ "NEVER_EXPIRE to set the expiry explicitly.",
);
}
const days = Number(raw);
@@ -202,7 +204,7 @@ function resolveDefaultExpiresAt(): string | null {
`[state_store] default_retention_days must be >= 0, got ${raw}. Set to 0
to disable expiry.`,
);
}
- return days === 0 ? null : toExpiresAt(days * MS_PER_DAY);
+ return days === 0 ? null : toExpiresAt(days * SECONDS_PER_DAY);
}
export function createCoordinatorClient(
@@ -277,7 +279,7 @@ export function createCoordinatorClient(
ti_id: tiId,
key,
value,
- expires_at: resolveExpiresAt(opts.retentionMs),
+ expires_at: resolveExpiresAt(opts.retention),
};
await rpc("SetTaskStateStore", "OKResponse", msg, () => undefined,
"throw");
},
diff --git a/ts-sdk/src/coordinator/operator-runner.ts
b/ts-sdk/src/coordinator/operator-runner.ts
index 6365770765a..bcb0cd47836 100644
--- a/ts-sdk/src/coordinator/operator-runner.ts
+++ b/ts-sdk/src/coordinator/operator-runner.ts
@@ -108,7 +108,7 @@ export function buildOperatorContext(deps:
OperatorContextDeps): OperatorContext
return {
type: "AwaitInputTask",
state: "awaiting_input",
- timeout: toIsoDuration(opts.timeoutSeconds),
+ timeout: toIsoDuration(opts.timeout),
next_method: EXECUTE_COMPLETE,
next_kwargs: opts.kwargs ?? {},
};
@@ -119,7 +119,7 @@ export function buildOperatorContext(deps:
OperatorContextDeps): OperatorContext
state: "deferred",
classpath: opts.classpath,
trigger_kwargs: opts.kwargs,
- trigger_timeout: toIsoDuration(opts.timeoutSeconds),
+ trigger_timeout: toIsoDuration(opts.timeout),
// `_defer_task` hands the trigger the task's queue only when
triggerer queues are enabled.
queue:
opts.queue !== undefined
diff --git a/ts-sdk/src/hitl/operator.ts b/ts-sdk/src/hitl/operator.ts
index dcea541ea94..6671f21360a 100644
--- a/ts-sdk/src/hitl/operator.ts
+++ b/ts-sdk/src/hitl/operator.ts
@@ -93,7 +93,7 @@ async function executeHITL(task: HITLTask<never>, op:
OperatorContext): Promise<
});
logs.info("Waiting for response", { task_id: ctx.taskId });
- return op.awaitInput({ timeoutSeconds: task.responseTimeout });
+ return op.awaitInput({ timeout: task.responseTimeout });
}
/**
diff --git a/ts-sdk/src/sdk/client-types.ts b/ts-sdk/src/sdk/client-types.ts
index 37c3b7b0e42..436921d808b 100644
--- a/ts-sdk/src/sdk/client-types.ts
+++ b/ts-sdk/src/sdk/client-types.ts
@@ -51,8 +51,8 @@ export interface SetXComOpts {
}
/**
- * Pass as {@link TaskStateStoreSetOpts.retentionMs} to store a task state
store
- * key with no expiry, regardless of the deployment default.
+ * Pass as {@link TaskStateStoreSetOpts.retention} to store a task state
+ * store key with no expiry, regardless of the deployment default.
*
* A symbol rather than a number, so a retention that reaches `Infinity` by
* accident (a division by zero, for example) is rejected instead of silently
@@ -63,14 +63,14 @@ export const NEVER_EXPIRE: unique symbol =
Symbol("apache-airflow-ts-sdk/NEVER_E
/** Options for a task state store write. */
export interface TaskStateStoreSetOpts {
/**
- * Milliseconds to retain the key, counted from the write; `0` expires it
- * immediately.
+ * How long to keep the key, in seconds, counted from the write; `0` expires
+ * it immediately.
*
* Omitting it follows the deployment's `[state_store]
default_retention_days`,
* which the coordinator passes to the runtime. Pass {@link NEVER_EXPIRE} to
* store the key with no expiry regardless of that default.
*/
- retentionMs?: number | typeof NEVER_EXPIRE;
+ retention?: number | typeof NEVER_EXPIRE;
}
/** Airflow Connection details returned by a task client. */
diff --git a/ts-sdk/src/sdk/client.ts b/ts-sdk/src/sdk/client.ts
index 6927fc47831..962638ee1c3 100644
--- a/ts-sdk/src/sdk/client.ts
+++ b/ts-sdk/src/sdk/client.ts
@@ -133,7 +133,7 @@ export interface TaskClient {
* let jobId = await store.get<string>("job_id");
* if (jobId == null) {
* jobId = await submit();
- * await store.set("job_id", jobId, { retentionMs: NEVER_EXPIRE });
+ * await store.set("job_id", jobId, { retention: NEVER_EXPIRE });
* }
* ```
*
@@ -157,9 +157,11 @@ export interface TaskStateStore {
/**
* Store a value, replacing any existing value for the key.
*
+ * `opts.retention` is how long to keep the key, in seconds.
+ *
* @throws `TypeError` when `key` is not a string, or `value` is null,
* undefined, NaN, or Infinity.
- * @throws `RangeError` when `key` is empty; when `retentionMs` is
+ * @throws `RangeError` when `key` is empty; when `retention` is
* negative, NaN, or Infinity; when the deployment's
* `[state_store] default_retention_days` is invalid; or when the resolved
* expiry would overflow the wire timestamp (year 9999).
diff --git a/ts-sdk/src/sdk/operator.ts b/ts-sdk/src/sdk/operator.ts
index 92468ac61ba..3184d0da55a 100644
--- a/ts-sdk/src/sdk/operator.ts
+++ b/ts-sdk/src/sdk/operator.ts
@@ -61,19 +61,19 @@ export interface OperatorContext {
fail(message: string): RuntimeRetryTask | RuntimeTaskState;
/** Skip `taskIds`, as `SkipMixin.skip` does. */
skip(taskIds: readonly string[]): Promise<void>;
- /** Park in `awaiting_input`. The next run calls `executeComplete`. */
+ /** Park in `awaiting_input`. The next run calls `executeComplete`.
`timeout` is in seconds. */
awaitInput(opts?: {
- timeoutSeconds?: number;
+ timeout?: number;
kwargs?: Record<string, JsonValue>;
}): RuntimeAwaitInputTask;
/**
* Defer to a Python trigger. The next run calls `executeComplete`. Without
a `queue`, the
- * trigger gets the task's queue when triggerer queues are enabled.
+ * trigger gets the task's queue when triggerer queues are enabled.
`timeout` is in seconds.
*/
defer(opts: {
classpath: string;
kwargs: Record<string, JsonValue>;
- timeoutSeconds?: number;
+ timeout?: number;
queue?: string | null;
}): RuntimeDeferTask;
}
diff --git a/ts-sdk/tests/coordinator/client.test.ts
b/ts-sdk/tests/coordinator/client.test.ts
index f5e864f6629..92fa03fcedf 100644
--- a/ts-sdk/tests/coordinator/client.test.ts
+++ b/ts-sdk/tests/coordinator/client.test.ts
@@ -165,7 +165,7 @@ describe("writes do not read a supervisor 404 as absence",
() => {
[
"taskStateStore.set",
"SetTaskStateStore",
- (c: TaskClient) => c.taskStateStore.set("k", 1, { retentionMs: 60_000 }),
+ (c: TaskClient) => c.taskStateStore.set("k", 1, { retention: 60 }),
],
[
"taskStateStore.delete",
@@ -446,7 +446,7 @@ describe("taskStateStore.set retention", () => {
vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
const { client: c, sent } = recordingClient({ type: "OKResponse", ok: true
});
- await c.taskStateStore.set("k", 1, { retentionMs: NEVER_EXPIRE });
+ await c.taskStateStore.set("k", 1, { retention: NEVER_EXPIRE });
expect(sent[0]).toEqual({
type: "SetTaskStateStore",
@@ -457,22 +457,22 @@ describe("taskStateStore.set retention", () => {
});
});
- it("sends now + retentionMs for a finite retentionMs", async () => {
+ it("sends now + retention for a finite retention", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
const { client: c, sent } = recordingClient({ type: "OKResponse", ok: true
});
- await c.taskStateStore.set("k", 1, { retentionMs: 60_000 });
+ await c.taskStateStore.set("k", 1, { retention: 60 });
expect(sent[0]).toMatchObject({ expires_at: "2026-01-01T00:01:00.000Z" });
});
- it("sends now (expire immediately) for retentionMs 0", async () => {
+ it("sends now (expire immediately) for retention 0", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
const { client: c, sent } = recordingClient({ type: "OKResponse", ok: true
});
- await c.taskStateStore.set("k", 1, { retentionMs: 0 });
+ await c.taskStateStore.set("k", 1, { retention: 0 });
expect(sent[0]).toMatchObject({ expires_at: "2026-01-01T00:00:00.000Z" });
});
@@ -535,15 +535,23 @@ describe("taskStateStore.set guards", () => {
it.each([
["negative", -1],
["NaN", Number.NaN],
- ])("rejects retentionMs=%s without sending a request", async (_label,
retentionMs) => {
+ ])("rejects retention=%s without sending a request", async (_label,
retention) => {
const { client: c, sent } = recordingClient();
- await expect(c.taskStateStore.set("k", 1, { retentionMs
})).rejects.toThrow(RangeError);
+ await expect(c.taskStateStore.set("k", 1, { retention
})).rejects.toThrow(RangeError);
+ expect(sent).toHaveLength(0);
+ });
+
+ it("reports an overflowing retention in seconds", async () => {
+ const { client: c, sent } = recordingClient();
+ await expect(c.taskStateStore.set("k", 1, { retention: 1e12
})).rejects.toThrow(
+ "a retention that ends 1000000000000s from now",
+ );
expect(sent).toHaveLength(0);
});
it("rejects an empty key without sending a request", async () => {
const { client: c, sent } = recordingClient();
- await expect(c.taskStateStore.set("", 1, { retentionMs: 60_000
})).rejects.toThrow(RangeError);
+ await expect(c.taskStateStore.set("", 1, { retention: 60
})).rejects.toThrow(RangeError);
expect(sent).toHaveLength(0);
});
});
@@ -577,7 +585,7 @@ describe("task state store key type guard", () => {
"set",
(c: TaskClient) => {
// @ts-expect-error exercising the runtime guard against an untyped
caller.
- return c.taskStateStore.set(undefined, 1, { retentionMs: 60_000 });
+ return c.taskStateStore.set(undefined, 1, { retention: 60 });
},
],
[
diff --git a/ts-sdk/tests/coordinator/operator-runner.test.ts
b/ts-sdk/tests/coordinator/operator-runner.test.ts
index 5aa5c76785e..17ba55717e0 100644
--- a/ts-sdk/tests/coordinator/operator-runner.test.ts
+++ b/ts-sdk/tests/coordinator/operator-runner.test.ts
@@ -157,13 +157,13 @@ describe("buildOperatorContext", () => {
it.each([
[undefined, null],
[30, "PT30S"],
- ])("parks with a timeout of %s as %s", (timeoutSeconds, timeout) => {
+ ])("parks with a timeout of %s as %s", (timeout, expected) => {
const { op } = makeContext();
- expect(op.awaitInput({ timeoutSeconds })).toEqual({
+ expect(op.awaitInput({ timeout })).toEqual({
type: "AwaitInputTask",
state: "awaiting_input",
- timeout,
+ timeout: expected,
next_method: "execute_complete",
next_kwargs: {},
});
@@ -178,7 +178,7 @@ describe("buildOperatorContext", () => {
it("defers to the trigger, resuming through execute_complete", () => {
const { op } = makeContext();
- expect(op.defer({ classpath: "pkg.Trigger", kwargs: { a: 1 },
timeoutSeconds: 5 })).toEqual({
+ expect(op.defer({ classpath: "pkg.Trigger", kwargs: { a: 1 }, timeout: 5
})).toEqual({
type: "DeferTask",
state: "deferred",
classpath: "pkg.Trigger",
diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts
index 163cb14471a..828fa7a32f4 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -449,7 +449,7 @@ describe("public API", () => {
>();
expectTypeOf<TaskClient["deleteVariable"]>().toEqualTypeOf<(key: string)
=> Promise<void>>();
expectTypeOf<TaskStateStoreSetOpts>().toEqualTypeOf<{
- retentionMs?: number | typeof NEVER_EXPIRE;
+ retention?: number | typeof NEVER_EXPIRE;
}>();
expectTypeOf<TaskClient["taskStateStore"]>().toEqualTypeOf<TaskStateStore>();
expectTypeOf<TaskStateStore["get"]>().toEqualTypeOf<
