sam-goodwin / eventual Goto Github PK
View Code? Open in Web Editor NEWBuild scalable and durable micro-services with APIs, Messaging and Workflows
Home Page: https://docs.eventual.ai
License: MIT License
Build scalable and durable micro-services with APIs, Messaging and Workflows
Home Page: https://docs.eventual.ai
License: MIT License
We currently have a mix of NodeJSFunction and custom bundling behavior. I think we should drop NodeJSFunction in favor of the eventual-bundle
script within @eventual/compiler
for the following reasons:
Add custom colors and logos to any activity or integration. Show them on the visualization(s).
Executions IDs contain /
which results in a 404 trying to find a handler in APIg.
Trying to URI encode doesn't seem to fix it.
abort
resolve
reject
let x = 0;
const cond1 = condition(() => x === 1);
event.on(() => {
eventual.reject(cond1);
});
if(await cond1) {
x = 2;
}
resolve
let x = 0;
const cond1 = condition(() => x === 1);
// when reject is called, cond1 returns a false response, even if the condition is met later.
event.on(() => {
eventual.resolve(cond1);
});
if(await cond1) {
x = 2; // when resolve is called, cond1 returns a true response, even if the condition is not met.
}
The alternative is using race and some local state.
let cancel = false;
event.on(() => {
eventual.resolve(cond1);
});
const cond1 = race(
condition(() => x === 1),
condtion(() => cancel === true).then(() => false)
);
if(await cond1) {
x = 2;
}
It should be possible to create an eventual.Service
with a RemovalPolicy that applies to all child resources.
undefined
[object Object]
✔ Execution id: sleepy/01GKP51QXFE2TEZSE1MWNZ29AG
ℹ undefined - WorkflowStarted- [object Object]
ℹ undefined - TaskStarted
ℹ undefined - TaskCompleted
ℹ undefined - TaskStarted
ℹ undefined - TaskCompleted
ℹ undefined - SleepScheduled
ℹ undefined - SleepScheduled
ℹ undefined - WorkflowCompleted
ℹ undefined - SleepScheduled
ℹ undefined - SleepScheduled
✔ Workflow complete
What is the purpose of the id
in the WorkflowTask
?
export interface WorkflowTask {
executionId: string;
id: string;
events: HistoryStateEvents[];
}
The service currently uses a ULID for this value - it's not deterministic, therefore we can't use it for anything valuable?
Step 1 in minimising CLI latency
I am little worried this won't scale with all of the Eventuals.
I think I would... group by seq and then use heuristics to determine 1. start 2. end and 3. other events
For Signals, just show them on the graph on their own (like start and end).
const groups = groupBy(events.filter(isHistoryEvent), event => event.seq);
const eventuals = Object.keys(groups).map(seq => {
const started = groups[seq].find(isScheduledEvent);
const ended = groups[seq].find(or(isCompletedEvent, isFailedEvent, isWorkflowTimedOut));
const other = groups[seq].filter(event => event !== started && event !== ended);
return { started, ended, other };
});
const otherEvents = events.filter(or(isSignalReceived, isWorkflowStarted, isWorkflowFailed, isWorkflowCompleted, isWorkflowTimedOut));
Originally posted by @thantos in #84 (comment)
The finishActivity
procedure prioritizes the execution history over submitting the SQS message which is slower. But this code can be run in parallel. No part of the orchestrator depends on the dynamodb state, so there is no need to add the overhead.
async function finishActivity(event: ActivityCompleted | ActivityFailed) {
await timed(metrics, ActivityMetrics.EmitEventDuration, () =>
executionHistoryClient.putEvent(request.executionId, event)
);
await timed(metrics, ActivityMetrics.SubmitWorkflowTaskDuration, () =>
workflowClient.submitWorkflowTask(request.executionId, event)
);
logActivityCompleteMetrics(isWorkflowFailed(event), event.duration);
}
If the Schedule Forwarder fails, we do not have a mechanism to retry forwarding and the EB Scheduler retry does not apply to function failures.
Will include:
workflow.startExecution should issue a Command which starts the workflow and returns the execution Id to the workflow code
History events are truncated.
payload: { amount: 1000 },
signalId: 'AddMoney',
timestamp: '2022-12-16T01:55:19.767Z'
},
{ type: 'TaskStarted', timestamp: '2022-12-16T01:55:19.909Z' },
{ type: 'TaskCompleted', timestamp: '2022-12-16T01:55:20.135Z' },
{ type: 'TaskStarted', timestamp: '2022-12-16T01:55:20.307Z' },
{ type: 'TaskCompleted', timestamp: '2022-12-16T01:55:20.349Z' },
{ type: 'TaskStarted', timestamp: '2022-12-16T01:55:50.327Z' },
... 75 more items
]
Should print to stdout on --json
, show a table otherwise, and support pagination.
Expect the CLI to return the executions in order with the last being the most visible to the user.
Order sees to be by workflow right now.
"endTime": "2022-12-13T04:11:50.327Z",
"error": "Timeout",
"message": "Workflow timed out",
"startTime": "2022-12-13T04:11:44.928Z",
"status": "FAILED"
},
{
"id": "slowWorkflow/##EVENTUAL##timedOut-01GM4VP15MF00JHBYJQKGWGQZ1-3",
"endTime": "2022-12-13T04:25:37.078Z",
"error": "Timeout",
"message": "Workflow timed out",
"startTime": "2022-12-13T04:25:31.559Z",
"status": "FAILED"
},
{
"id": "slowWorkflow/##EVENTUAL##timedOut-01GM4W0K7B8NSHRPDA17B6NMSZ-3",
"endTime": "2022-12-13T04:31:22.490Z",
"error": "Timeout",
"message": "Workflow timed out",
"startTime": "2022-12-13T04:31:17.104Z",
"status": "FAILED"
},
{
"id": "slowWorkflow/##EVENTUAL##timedOut-01GM4W31MVPNB72QT9CTB4QABQ-3",
"endTime": "2022-12-13T04:32:44.069Z",
"error": "Timeout",
"message": "Workflow timed out",
"startTime": "2022-12-13T04:32:38.350Z",
"status": "FAILED"
},
{
"id": "timedOut/01GKYJ0W6CT10B8PX5GRRVKWYK",
"endTime": "2022-12-10T17:41:18.064Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-10T17:41:12.530Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GKYJQHQAQ0Y43P9MF2VYAXKY",
"endTime": "2022-12-10T17:53:40.515Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-10T17:53:35.468Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GKYMZKP4VRGF1MRR12W869QD",
"endTime": "2022-12-10T18:33:01.873Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-10T18:32:56.776Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GKYNFZ7G4NGY809PDCCSD9G3",
"endTime": "2022-12-10T18:41:58.977Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-10T18:41:52.883Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GKYPM48VS8WTA0FWV15ZJYG1",
"endTime": "2022-12-10T19:01:42.995Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-10T19:01:37.697Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4RS8GPZX9M4XDD5KQ46Y8A",
"endTime": "2022-12-13T03:35:00.300Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T03:34:49.624Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4S12P6KK4YKY7FWW95YP5Q",
"endTime": "2022-12-13T03:39:15.523Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T03:39:05.799Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4SKS42MVCG8Z4W5PV46V4R",
"endTime": "2022-12-13T03:49:27.761Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T03:49:18.596Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4T7BYSW1AGK1FSNB1PSTTH",
"endTime": "2022-12-13T04:00:10.066Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T04:00:00.474Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4TWSXK50FEBNFFQPDB3S04",
"endTime": "2022-12-13T04:11:51.620Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T04:11:42.900Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4VP15MF00JHBYJQKGWGQZ1",
"endTime": "2022-12-13T04:25:37.701Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T04:25:29.526Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4W0K7B8NSHRPDA17B6NMSZ",
"endTime": "2022-12-13T04:31:23.684Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T04:31:15.693Z",
"status": "COMPLETE"
},
{
"id": "timedOut/01GM4W31MVPNB72QT9CTB4QABQ",
"endTime": "2022-12-13T04:32:45.475Z",
"result": {
"condition": true,
"signal": true,
"activity": true,
"workflow": true
},
"startTime": "2022-12-13T04:32:35.997Z",
"status": "COMPLETE"
}
The startWorkflow procedure currently runs the following operations in series:
Is it safe to do (3) before (1) or (2)?
The orchestrator doesn't rely on the ExecutionRecord or History event in DynamoDB to function correctly? Can't we write them last? Or all in parallel?
To go with Event
(#49), which has Singlecast semantics via execution.send
, we should support a broadcast semantic.
const ping = new Event<string>("ping");
const pong = new Event<string>("pong");
const worker = workflow(({ id: number }, { execution: { parentId } }) => {
const parent = controller.ref(parentId);
const sub = event.on(() => {
parent.send(pong, { ack: 1 });
});
await sleep(3 * 60);
if(id === 1) {
sub.dispose(); // unsubscribe;
}
})
const controller = workflow(() => {
const workers = [1,2,3].map((i) => worker.startWorkflow());
let resposnes = 0;
pong.on(() => { response++ });
// send to specific executions
ping.sendMany(workers, "hello");
await condition(() => responses === workers.len);
// send to anyone listening
ping.broadcast("hello");
})
sussmans@SurfacePro8:~/eventual$ AWS_PROFILE=sussman ./packages/@eventual/cli/bin/eventual.js start test-runner test-runner ./examples/lambda-test-runner/runtime/test-files/test-cases.json --tail
✔ Execution id: test-runner/01GM6TS7P83T039M83J69WKSFB
✖ No events at all. Check your execution id
sussmans@SurfacePro8:~/eventual$ AWS_PROFILE=sussman ./packages/@eventual/cli/bin/eventual.js start test-runner test-runner ./examples/lambda-test-runner/runtime/test-files/test-cases.json --tail
✔ Execution id: test-runner/01GM6TSRFJ0SMCMDSYM13QY87E
ℹ 2022-12-13T22:48:32.032Z - WorkflowStarted- {
target: {
name: 'arn:aws:lambda:us-east-1:983725774087:function:example-test-runner-testFunction483F4CBE-nQBI88y30Jfk',
type: 'lambda'
},
testCases: [ { input: [Object], expected: 3 } ]
}
ℹ 2022-12-13T22:48:32.146Z - TaskStarted
ℹ 2022-12-13T22:48:32.220Z - TaskCompleted
ℹ 2022-12-13T22:48:32.220Z - ActivityScheduled- runTestCase
ℹ 2022-12-13T22:48:32.392Z - ActivityCompleted- [object Object]
ℹ 2022-12-13T22:48:32.434Z - TaskStarted
ℹ 2022-12-13T22:48:32.453Z - WorkflowCompleted
ℹ 2022-12-13T22:48:32.453Z - TaskCompleted
✔ Workflow complete
{"passed":0,"failed":1,"errored":0,"results":[{"status":"fail","actual":{"0":51},"expected":3}]}
The activity token is currently base64 JSON object which contains the execution id and seq number of the activity.
This has a few issues:
Options:
Service internals are getting big. Lets split it.
Multiple workflows are allowed as of #33 :
export const workflow1 = workflow("worfklow1", () => ..);
export const workflow2 = workflow("worfklow2", () => ..);
The Workflow
Construct was renamed to Service
:
const myService = new eventual.Service(stack, "myService", {
entry: require.resolve("test-app-runtime/lib/my-workflow.js"),
});
So, what should we do with EventualApi
?
new eventual.EventualApi(stack, "api", {
services: [myWorkflow, openAccount],
});
Service
Construct and just orient the CLI tooling around a single service?ServiceHub
or ServiceManager
(something like that)?I am leaning towards (1) - orient everything around a single micro-service.
Support sending an event or a signal from the workflow with an activity token without starting a workflow.
note: not sure how to make this ergonomic
In an example I have...
const RequestApprovalEvent =
event<RequestApprovalEventPayload>("RequestApproval");
// auto approval for the human request approval event.
RequestApprovalEvent.on(async (event) => {
await requestApproval.complete({
activityToken: event.token,
result: { approve: true },
});
});
workflow(async () => {
await requestApproval({
price,
recommendation: decision,
symbol,
});
});
const requestApproval = activity(
"requestApproval",
async (event: Omit<RequestApprovalEventPayload, "token">) => {
return asyncResult<{ approve: boolean }>(async (token) => {
await RequestApprovalEvent.publish({ ...event, token });
});
}
);
Should be able to do this like...
const requestApprovalEventActivity = asyncEventActivity((token, data: Omit<RequestApprovalEventPayload, "token">) => ({
...data,
token
}));
workflow(async () => {
await requestApprovalEventActivity({
price,
recommendation: decision,
symbol,
});
});
This is one hell of a mouthful. We're gonna have to abstract this behind an api.
Originally posted by @sam-goodwin in #75 (comment)
Timeouts give the workflow author control over the durability of their own workflow. By setting timeouts, the author guarantees that they will get control back in the event of an unexpected issue.
myActivity.withTimeoutSeconds(100)
timeout(100, async () => { myActivity() })
Anything that blocks the workflow should be able to timeout.
Yes, in all cases timeouts are a mechanism that tasks the Orchestrator with waiting for time time and then interacting with the workflow. In all cases, the orchestrator interacts with the workflow by failing an Eventual (chain, call, etc). In the case of a workflow timeout, the top level Chain is failed. In the case of an activity, the activity call being waited on is failed.
Anything can be given a timeout using the sleepFor/sleepUntil calls.
Note:
Promise.any/race
is not yet implemented. #62
const complexThing = chain(async () => {
await act1();
await act2();
});
Promise.race(complexThing(), sleepFor()); // or Promise.any()
const act1 = activity("act1", { timeoutSeconds: 1000 }, () => {}); // times out after 1000 seconds
// times out after 1000 seconds
const wf1 = workflow("wf1", { timeoutSeconds: 1000 }, () => {
await expectSignal("mySignal", { timeoutSeconds: 100 }); // times out after 100 seconds if signal is not recieved
await condition({ timeoutSeconds: 100 }}, () => x === y); // times out after 100 seconds if the condition has yet to be met
await act1() // times out after 1000 seconds if does not return.
await act1.withOptions({ timeoutSeconds: 100 })(); // overrides the timeout to 100 seconds if the activity has not completed
});
// runs until completion or error, no timeout
const wf2 = workflow("wf2", () => {
const child = await wf1(); // will fail after 1000 seconds
const child2 = await wf1.withOptions({ childTimeoutSeconds: 100 })(); // will fail after 100 seconds
});
Note: see below (Workflow timeout vs Child Workflow) for details about child workflow timeout options.
An interesting case is the difference between the workflow having a top level timeout and a workflow timing out on the operation of it's child. Under the current assumption that a failed workflow does not terminate it's children (workflows it started) the timeout of a child workflow eventual on the parent should not fail the child.
In the below example, if the child continues to run for 100 seconds without returning a completed message, an error will be thrown from the await (reject). The child will continue to run, may even success, and the parent, in theory, could still query for it's result again.
const childWf = workflow("childWf", () => { ... long running things... });
const parentWf = workflow("parentWf", () => {
await childWf.withOptions({ timeoutSeconds: 100 })(); // throw an error to the parent after 100 seconds, child will continue
});
If the child itself times out, it will also throw an error in the parent.
const childWf = workflow("childWf", { timeoutSeconds: 100 }, () => { ... long running things... });
const parentWf = workflow("parentWf", () => {
await childWf(); // throw an error to the parent after 100 seconds, child will is failed with timeout
});
Note: if the child is never awaited on a timeout will be ignored
const childWf = workflow("childWf", { timeoutSeconds: 100 }, () => { ... long running things... });
const parentWf = workflow("parentWf", () => {
const child = childWf(); // child times out after 100 seconds, nothing happens here
...doOtherWork...
await child; // throws
});
We will allow the configuration of two different timeouts on the child workflow to handle the ambiguity.
ScheduleChildWorkflow
event) to the it being resolved or failed.const childWf = workflow("childWf", { timeoutSeconds: 1000 }, () => { ... long running things... });
const parentWf = workflow("parentWf", () => {
await childWf(); // the execution will timeout after 1000 seconds
await childWf.withOptions({ executionTimeoutSeconds: 100 })(); // the execution will timeout after 100 seconds
await childWf.withOptions({ childTimeoutSeconds: 100 })(); // the execution will timeout after 1000 seconds, but the parent will fail the eventual after 100 seconds
});
Should be able to cancel a running workflow execution
await cancelWorkflow(executionId, reason);
// cancels itself (?)
throw new CancelWorkflow(reason);
const execution = workflow.startExecution();
await execution.cancel(reason);
const execution = workflow();
await execution.cancel(reason);
As a developer/workflow author, I want to create activities that run for indefinite amounts of time, involve human interaction, invoke other services, or wait for the result of outside actions. I should be able to ensure inconsistencies and fault are recoverable from. I should be able to use the service to support idempotency of partial failures.
workflow(() => {
const result = await act1();
});
act1 = activity<{ result: string }>({
heartbeat: { seconds: number },
timeout: { seconds: number }
}, (context: Context): { result: string } | AsyncToken => {
...doSomeWork...
await sendToQueue({ token: context.activity.token });
return makeAsync();
})
// lambda function consuming the queue
const workflowClient = new WorkflowClient();
export const handler = async (event) => {
Promise.all(event.Records.map(async (record) => {
const payload = JSON.parse(record.body);
// complete the activity with a payload
await workflowClient.completeActivity<typeof act1>(
payload.token,
{ result: "done"}
);
// or fail
await workflowClient.failActivity(payload.token, {result: "done"});
}));
}
workflow(() => {
const result = await act1();
});
act1 = activity<{ result: string }>({
heartbeat: { seconds: 20 },
timeout: { seconds: 160 }
}, (context: Context): { result: string } | AsyncToken => {
...doSomeWork...
await sendToQueue({ token: context.activity.token });
await sendHeartbeat();
return makeAsync();
})
// lambda function consuming the queue
const workflowClient = new WorkflowClient();
export const handler = async (event) => {
Promise.all(event.Records.map(async (record) => {
const payload = JSON.parse(record.body);
while(true) {
// some long process
await workflowClient.heartbeatActivity(payload.token);
}
// complete the activity with a payload
await workflowClient.completeActivity<typeof act1>(
payload.token,
{ result: "done"}
);
}));
}
workflow(() => {
const act = act1();
act.onHeartbeat(async ({ i: 100 }) => {
await reportProgress(i);
});
});
const reportProceess = activity(...);
const act1 = activity<{ result: string }, { i: 100 } | undefined>({
heartbeat: { seconds: 20 },
timeout: { seconds: 160 }
}, (context: Context): { result: string } | AsyncToken => {
...doSomeWork...
await sendToQueue({ token: context.activity.token, start: context.checkpoint });
// should this be on the context to be typed?
await sendHeartbeat();
return makeAsync();
})
// lambda function consuming the queue
const workflowClient = new WorkflowClient();
export const handler = async (event) => {
Promise.all(event.Records.map(async (record) => {
const payload = JSON.parse(record.body);
const items = [...];
const start = event.start ?? 0;
for(const i of items.slice()) {
// some long process
await workflowClient.heartbeatActivity<typeof act1>(
payload.token,
{ i }
);
}
// complete the activity with a payload
await workflowClient.completeActivity<typeof act1>(
payload.token,
{ result: "done"}
);
}));
}
This is bloating our logs and i wonder if it's a problem?
ts-jest[ts-compiler] (WARN) Unable to process '/home/runner/work/eventual/eventual/packages/@eventual/core/lib/cjs/error.js', falling back to original file content. You can also configure Jest config option `transformIgnorePatterns` to ignore /home/runner/work/eventual/eventual/packages/@eventual/core/lib/cjs/error.js from transformation or make sure that `outDir` in your tsconfig is neither `''` or `'.'`
Execution Id | Status | Start Time | End Time | Parent id |
---|---|---|---|---|
console.log
calls in a customer's workflow should only emit logs to the log stream the first time they are run. We need to monkey-patch them to use the history to dedupe behavior.
This work can probably be done in the orchestrator shim
A useState
like call.
workflow(() => {
const state = new State<"IN_PROGRESS" | "SUCCESS">("IN_PROGRESS");
signal.on(() => {
state.update("SUCCESS");
});
await condition(() => state.value === "IN_PROGRESS");
return "DONE";
})
Or
Temporal has a concept called a "trigger" which is a promise that can be resolved. Entity/State would be similar, but support any value and could be "resolved"/set multiple times.
This would works well with conditions. Conditions do not trigger Workflow Runs on their own, they need to be triggered by something. State will trigger a Complete/Update event when the status changes, in order of the status change.
We could also make shared entities and complex entities, which get closer to DurableEntities/actors.
Lambda - EMF to CloudWatch
SQS
Dynamo
Introduces:
import { workflow, WorkflowHandler } from "@eventual/core";
declare class Event<Payload = void> {
constructor(public readonly id: string) {}
on(handler: (payload: Payload) => Promise<void> | void): Promise<void>;
}
type EventPayload<E extends Event<any>> = E extends Event<infer P> ? P : never;
declare function waitForEvent<E extends Event<any>>(
event: E,
opts: { timeoutSeconds: number }
): Promise<EventPayload<E>>;
declare function condition(
predicate: () => boolean,
opts: { timeoutSeconds: number }
): Promise<void>;
const event = new Event<number>("event");
const doneEvent = new Event("done");
declare module "@eventual/core" {
interface Workflow<F extends WorkflowHandler = WorkflowHandler> {
ref(executionId: string): ExecutionRef;
startExecution(input: Parameters<F>[0]): Promise<ExecutionRef>;
}
interface ExecutionRef {
send<E extends Event<any>>(event: E, payload: EventPayload<E>): void;
}
}
/**
* the parent workflow uses thr `waitForEvent` function to block and wait for events from it's child workflow.
*/
export const workflow1 = workflow("workflow1", async () => {
const child = await workflow2.startExecution({ name: "child" });
while (true) {
const n = await waitForEvent(event);
console.log(n);
if (n > 10) {
child.send(doneEvent, undefined);
break;
}
child.send(event, n + 1);
}
return "done";
});
/**
* The child workflow shows a different way of having events using handlers, conditions, and local state.
*/
export const workflow2 = workflow(
"workflow2",
async (input: { name: string }, { execution: { parentId } }) => {
let block = false;
let done = false;
let last = 0;
if (!parentId) {
throw new Error("I need an adult");
}
console.log(`Hi, I am ${input.name}`);
const parent = workflow1.ref(parentId);
await event.on((n) => {
last = n;
block = false;
});
await doneEvent.on(() => {
done = true;
block = false;
});
while (!done) {
parent.send(event, last + 1);
block = true;
await condition(() => !block);
}
return "done";
}
);
EB Schedules send events to the Scheduler Forwarder lambda. This may fail after X retries and Y time to a DLQ. We need to durably handle these DLQ messages and have metrics/alarms when messages land in there.
We should be conservative with the number of resources we provision. Is there a good reason to partition locks and history?
Promise.
@eventual/integrations-slack
that integrates the Slack API as activities and webhook apis #89The workflow may need to make use of values that are generated immediately by a command's execution, but do not require the action to be "complete".
A parent workflow MUST be able to manipulate the child, like a parent/child thread (fork). ex: send signals to the child, cancel the child, and await it's result (join).
const wf = workflow("wf1", () => {});
const wf2 = workflow("wf2", () => {
const child = wf();
child.sendSignal("MySignal"); // error: how does send signal know the ID?
sendSignal(child.executionId, "MySignal"); // error: executionId property cannot be known here as scheduleWorkflow and sendSignal are executed in parallel
await child();
})
In the above examples, the commands output would look like
ScheduleWorkflow
SendSignal // how do I get the execution id?
SendSignal // how do I get the execution id?
This is what Temporal does, they have
startChild
andexecuteChild
that do the same asstartWorkflow
andexecuteWorkflow
below. https://docs.temporal.io/typescript/workflows#child-workflows
const wf2 = workflow("wf2", () => {
const childHandle = await wf.startExecution(); // returns a execution handle with the ID.
const child = childHandle.result() // returns a promise related to the end of the workflow.
childHandle.sendSignal("MySignal");
sendSignal(childHandle.executionId, "MySignal");
return await child; //waits for completion and returns result
})
startWorkflow
and executeWorkflow
)startWorkflow
returns a handle with functions and propertiesexecuteWorkflow
just returns an awaitable that awaits the results.startWorkflow
API must be usedPros: we know it would work
Cons: kind of ugly, might need to take multiple loops of awaiting and resolving
Currently the execution name of the workflow is dynamically generated. If we made the name deterministically generated, the executionId could be generated.
Pros: simple?
Cons: Race condition if the signal is sent before the child workflow is started
const wf2 = workflow("wf2", () => {
const child = wf(); // returns a execution handle with the ID and a "Promise/Eventual"
child.sendSignal("MySignal");
sendSignal(child.executionId, "MySignal");
console.log(child.executionId) // { ref: "ExecutionId", seq: 0 }
return await child; // returns result
})
executionId
can return Tokens (like CDK) which resolve to a property we expect to exist after a command is called.Pros: cleanest solution, enforces dependency order of commands with best attempt concurrency
Cons: Most complex implementation, may limit command execution parallelism (in rare cases), need to carefully resolve tokens
Pros: no work
Cons: lose a strong use case
It should be possible to control activities from the calling workflow or from other workflows.
api.any(async () => {
// or event handler or workflow
await cancelActivity(activityToken, reason);
await failActivity(activityToken, error, message);
await completeActivity(activityToken, result);
})
const myAct = activity(...);
workflow(() => {
// start activity
const actInstance = myAct();
// cancel, error, or complete the workflow
await actInstance.cancel(reason);
await actInstance.fail(new Error()); // or actInstance.fail(error, message);
await actInstance.complete(result);
await actInstance;
});
RE: #89 (comment)
We use a proxy in the slack integration (#89) to remotely execute a call within a durable activity instead of within an orchestrator. This pattern will likely be useful for many different use-cases, especially integrations, we should extract it into core.
Our API clients require all environment variables even when a user only need a subset of the API operations
Our configureXYZ metrics reflect what they ideally need but not what they actually need, causing errors in the development API
public configureRecordHistory(func: Function) {
this.grantRecordHistory(func);
addEnvironment(func, {
[ENV_NAMES.EXECUTION_HISTORY_BUCKET]: this.history.bucketName,
});
}
Yes, recording history only needs the bucket but you can't construct the API without the WORKFLOW_QUEUE_URL etc.
As an example, have this recurring error currently occuring:
Executions failed: \nremindMe/01GM3GQ6A6MCZ65XEF3G0GMR78: Error: no such workflow with name 'remindMe'
Presumably since the code was disappeared after a sleep/schedule was put in place.
In fatal cases such as these, with no hope of recovery, we'd want retries to be abandoned
Data returned from activities and workflows may be larger than can we stored in dynamo and sqs.
new Rule().addTarget(new EventualServiceTarget(service));
Allow forwarding events from another event bus to the Service.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.