diff --git a/CHANGELOG.md b/CHANGELOG.md index fa7c3a65..9ad92e31 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ### New +- Expose readonly `ActivityContext.name` and `ActivityContext.version` from the activity request, + preserving the requested version even when an unversioned implementation handles it. Existing + two-argument context construction remains supported, with empty name/version defaults. - Add optional version arguments to orchestrator/activity registrations in core and test workers, dispatch by recorded/request version, and preserve versions in the in-memory backend. Same-name versions coexist; unversioned-only registrations retain the .NET-compatible fallback. diff --git a/packages/durabletask-js/src/task/context/activity-context.ts b/packages/durabletask-js/src/task/context/activity-context.ts index b06d8265..05f4b378 100644 --- a/packages/durabletask-js/src/task/context/activity-context.ts +++ b/packages/durabletask-js/src/task/context/activity-context.ts @@ -4,10 +4,35 @@ export class ActivityContext { private _orchestrationId: string; private _taskId: number; + private readonly _name: string; + private readonly _version: string; - constructor(orchestrationId: string, taskId: number) { + /** + * @param orchestrationId - The ID of the orchestration that scheduled this activity. + * @param taskId - The scheduled task ID. + * @param name - The logical activity request name. Defaults to empty for legacy two-argument callers. + * @param version - The requested activity version. Defaults to empty when not supplied. + */ + constructor(orchestrationId: string, taskId: number, name: string = "", version: string = "") { this._orchestrationId = orchestrationId; this._taskId = taskId; + this._name = name; + this._version = version ?? ""; + } + + /** + * Gets the logical activity name from the request, not the implementation function name. + */ + get name(): string { + return this._name; + } + + /** + * Gets the requested activity version, or an empty string when none was supplied. + * Preserves the request's casing and version even when an unversioned registration handles it. + */ + get version(): string { + return this._version; } get orchestrationId(): string { diff --git a/packages/durabletask-js/src/worker/activity-executor.ts b/packages/durabletask-js/src/worker/activity-executor.ts index 65efbe21..4c4a84b4 100644 --- a/packages/durabletask-js/src/worker/activity-executor.ts +++ b/packages/durabletask-js/src/worker/activity-executor.ts @@ -33,7 +33,7 @@ export class ActivityExecutor { // Log activity start (EventId 603) WorkerLogs.activityStarted(this._logger, orchestrationId, name); - const ctx = new ActivityContext(orchestrationId, taskId); + const ctx = new ActivityContext(orchestrationId, taskId, name, version); try { // Deserialize the input inside the try-catch so that malformed JSON diff --git a/packages/durabletask-js/test/activity_executor.spec.ts b/packages/durabletask-js/test/activity_executor.spec.ts index 9a289525..d66e1aff 100644 --- a/packages/durabletask-js/test/activity_executor.spec.ts +++ b/packages/durabletask-js/test/activity_executor.spec.ts @@ -15,7 +15,56 @@ const testLogger = new NoOpLogger(); const TEST_INSTANCE_ID = "abc123"; const TEST_TASK_ID = 42; +describe("Activity Context", () => { + it("keeps the two-argument constructor compatible with empty metadata", () => { + const ctx = new ActivityContext(TEST_INSTANCE_ID, TEST_TASK_ID); + expect(ctx).toMatchObject({ + orchestrationId: TEST_INSTANCE_ID, + taskId: TEST_TASK_ID, + name: "", + version: "", + }); + }); + + it("normalizes a null version supplied by JavaScript callers", () => { + const ctx = Reflect.construct(ActivityContext, [TEST_INSTANCE_ID, TEST_TASK_ID, "LogicalActivity", null]); + expect(ctx).toMatchObject({ + orchestrationId: TEST_INSTANCE_ID, + taskId: TEST_TASK_ID, + name: "LogicalActivity", + version: "", + }); + }); +}); + describe("Activity Executor", () => { + it.each([ + ["exact version", "2.0.0", "2.0.0"], + ["case-insensitive lookup", "2.0.0-rc", "2.0.0-RC"], + ["unversioned registration fallback", undefined, "2.0.0"], + ["missing version", undefined, undefined], + ["explicit empty version", undefined, ""], + ])("populates request metadata for %s", async (_scenario, registrationVersion, requestVersion) => { + let actualContext: ActivityContext | undefined; + function implementation(ctx: ActivityContext) { + actualContext = ctx; + return "done"; + } + const registry = new Registry(); + registry.addNamedActivity("LogicalActivity", implementation, registrationVersion); + const executor = new ActivityExecutor(registry, testLogger); + + await expect( + executor.execute(TEST_INSTANCE_ID, "LogicalActivity", TEST_TASK_ID, undefined, requestVersion), + ).resolves.toBe('"done"'); + expect(actualContext).toMatchObject({ + orchestrationId: TEST_INSTANCE_ID, + taskId: TEST_TASK_ID, + name: "LogicalActivity", + version: requestVersion ?? "", + }); + }); + it("should validates activity function input population", async () => { const testActivity = (ctx: ActivityContext, testInput: any) => { // return all activity inputs back as the output diff --git a/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts b/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts index 333746be..d840ecb4 100644 --- a/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts +++ b/packages/durabletask-js/test/versioned-dispatch-grpc.spec.ts @@ -3,7 +3,14 @@ import * as grpc from "@grpc/grpc-js"; import { Empty } from "google-protobuf/google/protobuf/empty_pb"; -import { NoOpLogger, TaskHubGrpcWorker, VersionMatchStrategy, VersionFailureStrategy, VersioningOptions } from "../src"; +import { + ActivityContext, + NoOpLogger, + TaskHubGrpcWorker, + VersionMatchStrategy, + VersionFailureStrategy, + VersioningOptions, +} from "../src"; import * as pb from "../src/proto/orchestrator_service_pb"; import * as stubs from "../src/proto/orchestrator_service_grpc_pb"; import * as ph from "../src/utils/pb-helper.util"; @@ -121,6 +128,35 @@ describe("Version dispatch over local gRPC", () => { return { result: action?.getResult()?.getValue(), failure: action?.getFailuredetails() }; } + it.each<[string, string | undefined]>([ + ["LogicalActivity", "2.0.0"], + ["LogicalActivity", undefined], + ["LogicalActivity", ""], + ["VersionedActivity", "2.0.0-RC"], + ])("exposes activity request metadata for %s / %s over gRPC", async (name, version) => { + let actualContext: ActivityContext | undefined; + function implementation(ctx: ActivityContext) { + actualContext = ctx; + return "done"; + } + const stream = await start( + (worker) => { + worker.addNamedActivity("LogicalActivity", implementation); + worker.addNamedActivity("VersionedActivity", implementation, "2.0.0-rc"); + }, + { defaultVersion: "worker-default" }, + ); + const response = completion(await send(stream, "activity", name, version)); + expect(response.failure).toBeUndefined(); + expect(response.result).toBe('"done"'); + expect(actualContext).toMatchObject({ + orchestrationId: "instance", + taskId: 1, + name, + version: version ?? "", + }); + }); + it("dispatches same-name versions, legacy fallback, and unknown versions without cross-version execution", async () => { const calls: string[] = []; const stream = await start((worker) => {