Add coalesce example #281
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| import * as restate from "@restatedev/restate-sdk"; | ||
|
|
||
| interface TargetWorkflowInput {} | ||
| interface TargetWorkflowOutput { | ||
| key: string; | ||
| } | ||
|
|
||
| const targetWorkflow = restate.workflow({ | ||
| name: "targetWorkflow", | ||
| handlers: { | ||
| run: async ( | ||
| ctx: restate.WorkflowContext, | ||
| input: TargetWorkflowInput, | ||
| ): Promise<TargetWorkflowOutput> => { | ||
| // simulate some work happening | ||
| await ctx.sleep(5_000); | ||
|
|
||
| return { key: ctx.key }; | ||
| }, | ||
| }, | ||
| }); | ||
| export type TargetWorkflow = typeof targetWorkflow; | ||
| const TargetWorkflow: TargetWorkflow = { name: "targetWorkflow" }; | ||
|
|
||
| interface Target { | ||
| service: string; | ||
| handler: string; | ||
| key?: string; | ||
| } | ||
|
|
||
| // supervisor watches an invocation (must be from a workflow or with an idempotency key) for its completion and then calls back | ||
| const supervisor = restate.service({ | ||
| name: "supervisor", | ||
| handlers: { | ||
| watch: async ( | ||
| ctx: restate.Context, | ||
| input: { invocationId: restate.InvocationId; target: Target }, | ||
| ): Promise<void> => { | ||
| const done = () => | ||
| ctx.genericSend({ | ||
| service: input.target.service, | ||
| method: input.target.handler, | ||
| key: input.target.key, | ||
| parameter: input.invocationId, | ||
| inputSerde: restate.serde.json, | ||
| }); | ||
|
|
||
| try { | ||
| await ctx.attach(input.invocationId, restate.serde.binary); | ||
| } catch (e) { | ||
| if (e instanceof restate.TerminalError) { | ||
| done(); | ||
| } | ||
| throw e; | ||
| } | ||
| done(); | ||
| }, | ||
| }, | ||
| }); | ||
|
|
||
| type Supervisor = typeof supervisor; | ||
| const Supervisor: Supervisor = { name: "supervisor" }; | ||
|
|
||
| interface CoalesceState { | ||
| invocationId: restate.InvocationId; | ||
| count: number; | ||
| } | ||
|
|
||
| const coalesce = restate.object({ | ||
| name: "coalesce", | ||
| handlers: { | ||
| submit: restate.handlers.object.shared( | ||
| async ( | ||
| ctx: restate.ObjectSharedContext<CoalesceState>, | ||
| input: TargetWorkflowInput, | ||
| ): Promise<TargetWorkflowOutput> => { | ||
| let invocationId = await ctx.get("invocationId"); | ||
| if (invocationId == null) { | ||
| // there might be no in flight invocation, run the slow path to be sure | ||
| invocationId = await ctx | ||
| .objectClient(Coalesce, ctx.key) | ||
| .submitLocked(input); | ||
| } | ||
|
|
||
| return ctx.attach(invocationId); | ||
| }, | ||
| ), | ||
| submitLocked: async ( | ||
| ctx: restate.ObjectContext<CoalesceState>, | ||
| input: TargetWorkflowInput, | ||
| ): Promise<restate.InvocationId> => { | ||
| let invocationId = await ctx.get("invocationId"); | ||
| const count = (await ctx.get("count")) ?? 0; | ||
|
|
||
| if (invocationId !== null) { | ||
| // we lost the race | ||
| return invocationId; | ||
| } | ||
|
|
||
| // there is no in flight invocation, create one | ||
| const workflowID = `${ctx.key}-${count}`; | ||
| const handle = ctx | ||
| .workflowSendClient(TargetWorkflow, workflowID) | ||
| .run(input); | ||
|
There was a problem hiding this comment. Thank you for putting together this example. I guess here we could also do |
||
| invocationId = await handle.invocationId; | ||
|
|
||
| ctx.serviceSendClient(Supervisor).watch({ | ||
| invocationId, | ||
| target: { | ||
| service: "coalesce", | ||
| handler: "onCompletion", | ||
| key: ctx.key, | ||
| }, | ||
| }); | ||
|
|
||
| ctx.set("invocationId", invocationId); | ||
| ctx.set("count", count + 1); | ||
|
|
||
| return invocationId; | ||
| }, | ||
| onCompletion: async ( | ||
| ctx: restate.ObjectContext<CoalesceState>, | ||
| invocationId: restate.InvocationId, | ||
| ) => { | ||
| if ((await ctx.get("invocationId")) == invocationId) { | ||
| ctx.clear("invocationId"); | ||
| } | ||
| }, | ||
| }, | ||
| }); | ||
|
|
||
| type Coalesce = typeof coalesce; | ||
| const Coalesce: Coalesce = { name: "coalesce" }; | ||
|
|
||
| restate | ||
| .endpoint() | ||
| .bind(targetWorkflow) | ||
| .bind(supervisor) | ||
| .bind(coalesce) | ||
| .listen(); | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can this be improved to explain how the behavior here is different from just using an idempotency key?