Implement workflows and durable promises (#40)

* Implement workflows

* Add workflow example to codegen

* Update handler.go

Co-authored-by: Francesco Guardiani <[email protected]>

* Update facilitators.go

Co-authored-by: Francesco Guardiani <[email protected]>

* Update facilitators.go

Co-authored-by: Francesco Guardiani <[email protected]>

* Update handler.go

Co-authored-by: Francesco Guardiani <[email protected]>

* Update router.go

Co-authored-by: Francesco Guardiani <[email protected]>

* Fix awakeable holder race

---------

Co-authored-by: Francesco Guardiani <[email protected]>

context.go

@@ -27,7 +27,7 @@ type Context interface {
 	inner() *state.Context
 }
 
-// ObjectContext is an extension of [Context] which is passed to shared-mode Virtual Object handlers,
+// ObjectSharedContext is an extension of [Context] which is passed to shared-mode Virtual Object handlers,
 // giving read-only access to a snapshot of state.
 type ObjectSharedContext interface {
 	Context
@@ -40,3 +40,18 @@ type ObjectContext interface {
 	ObjectSharedContext
 	exclusiveObject()
 }
+
+// WorkflowSharedContext is an extension of [ObjectSharedContext] which is passed to shared-mode Workflow handlers,
+// giving read-only access to a snapshot of state.
+type WorkflowSharedContext interface {
+	ObjectSharedContext
+	workflow()
+}
+
+// WorkflowContext is an extension of [WorkflowSharedContext] and [ObjectContext] which is passed to Workflow 'run' handlers,
+// giving mutable access to state.
+type WorkflowContext interface {
+	WorkflowSharedContext
+	ObjectContext
+	runWorkflow()
+}

examples/codegen/main.go

@@ -118,10 +118,37 @@ func (c counter) Watch(ctx restate.ObjectSharedContext, req *helloworld.WatchReq
 	return &helloworld.GetResponse{Value: next}, nil
 }
 
+type workflow struct {
+	helloworld.UnimplementedWorkflowServer
+}
+
+func (workflow) Run(ctx restate.WorkflowContext, _ *helloworld.RunRequest) (*helloworld.RunResponse, error) {
+	restate.Set(ctx, "status", "waiting")
+	_, err := restate.Promise[restate.Void](ctx, "promise").Result()
+	if err != nil {
+		return nil, err
+	}
+	restate.Set(ctx, "status", "finished")
+	return &helloworld.RunResponse{Status: "finished"}, nil
+}
+
+func (workflow) Finish(ctx restate.WorkflowSharedContext, _ *helloworld.FinishRequest) (*helloworld.FinishResponse, error) {
+	return nil, restate.Promise[restate.Void](ctx, "promise").Resolve(restate.Void{})
+}
+
+func (workflow) Status(ctx restate.WorkflowSharedContext, _ *helloworld.StatusRequest) (*helloworld.StatusResponse, error) {
+	status, err := restate.Get[string](ctx, "status")
+	if err != nil {
+		return nil, err
+	}
+	return &helloworld.StatusResponse{Status: status}, nil
+}
+
 func main() {
 	server := server.NewRestate().
 		Bind(helloworld.NewGreeterServer(greeter{})).
-		Bind(helloworld.NewCounterServer(counter{}))
+		Bind(helloworld.NewCounterServer(counter{})).
+		Bind(helloworld.NewWorkflowServer(workflow{}))
 
 	if err := server.Start(context.Background(), ":9080"); err != nil {
 		slog.Error("application exited unexpectedly", "err", err.Error())