Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add Nexus APIs to the Temporal Go SDK #89

Merged
merged 4 commits into from
May 7, 2024
+369 −0
Merged

Add Nexus APIs to the Temporal Go SDK #89

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
49b1fc6
Add proposal for adding Nexus APIs to the Temporal Go SDK
bergundy Apr 23, 2024
5b20f2b
Address some review comments
bergundy Apr 24, 2024
68bb78f
Add service concept
bergundy Apr 29, 2024
7c1b204
NexusClient
bergundy Apr 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 12 additions & 53 deletions nexus/sdk-go.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,38 +237,10 @@ SDK's selectors.

> NOTE: In the future, as the Go SDK adds support for typed futures, we will add a strongly typed variant of this API.

```go
package temporalnexus

type Endpoint interface {
// contains filtered or unexported methods
}

// NewNamedEndpoint creates an with targeted by name as registered in the endpoint registry. If the endpoint is renamed
// calls refering to the old name will fail.
func NewNamedEndpoint(name string) Endpoint

// START: Potential future

// NewEndpointFromID creates an endpoint targeted by a stable endpoint ID from the endpoint registry.
func NewEndpointFromID(id string) Endpoint

// NewEndpointFromSelectors creates an endpoint that is resolved by applying the given selectors.
// For example {"env" "prod", "region": "us-east-1"}.
func NewEndpointFromSelectors(selectors map[string]string) Endpoint

// END: Potential future
```

```go
// NexusOperationOptions are options for starting a Nexus Operation from a Workflow.
type NexusOperationOptions struct {
ScheduleToCloseTimeout time.Duration
// The endpoint to send this operation request to. Optional. If not provided the endpoint will be resolved in
// the worker based on the given worker options NexusEndpointResolver function.
// It is generally recommended to keep workflow code enviroment agnostic and only set this from a workflow in
// ambiguous cases.
Endpoint temporalnexus.Endpoint
}

// NexusOperationExecution is the result of [NexusOperationFuture.GetNexusOperationExecution].
Expand All @@ -288,9 +260,15 @@ type NexusOperationFuture interface {
GetNexusOperationExecution() Future
}

// ExecuteNexusOperation executes a Nexus Operation.
// The operation argument can be a string, a [nexus.Operation] or a [nexus.OperationReference].
func ExecuteNexusOperation(ctx Context, service string, operation any, input any, options NexusOperationOptions) NexusOperationFuture
// NexusClient is a client for executing Nexus Operations from a workflow.
type NexusClient interface {
// ExecuteOperation executes a Nexus Operation.
// The operation argument can be a string, a [nexus.Operation] or a [nexus.OperationReference].
ExecuteOperation(ctx Context, operation any, input any, options NexusOperationOptions) NexusOperationFuture
}

// Create a [NexusClient] from a service name and an endpoint name.
func NewNexusClient(service, endpoint string) NexusClient
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider reversing the order to endpoint, service.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also considering adding context and an options struct to future proof this API.

```

**Usage**:
Expand All @@ -301,15 +279,12 @@ import (
"time"

"go.temporal.io/sdk/workflow"
"go.temporal.io/sdk/temporalnexus"
)

func MyCallerWorkflow(ctx workflow.Context) (MyOutput, error) {
fut := workflow.ExecuteNexusOperation(ctx, "payments", "start-transaction", MyInput{ID: "tx-deadbeef"}, workflow.NexusOperationOptions{
client := workflow.NewNexusClient("payments", "prod-payments")
fut := client.ExecuteOperation(ctx, "start-transaction", MyInput{ID: "tx-deadbeef"}, workflow.NexusOperationOptions{
ScheduleToCloseTimeout: time.Hour,
// Optional, if not provided the endpoint will be resolved in the worker based on the given worker
// options.
Endpoint: temporalnexus.NewNamedEndpoint("prod-payments"),
})
var exec workflow.NexusOperationExecution
_ = fut.GetNexusOperationExecution().Get(ctx, &exec)
Expand All @@ -321,30 +296,14 @@ func MyCallerWorkflow(ctx workflow.Context) (MyOutput, error) {

> NOTE: To cancel a Nexus Operation, cancel the context used to execute it.

### Endpoint Registration in the Worker

To avoid injecting configuration into the workflow - which is inherently unsafe and could break determinism - the SDK
provides a worker option to resolve a service to an endpoint.

> NOTE: This could also be implemented in an interceptor but for consistency with all SDKs, including ones where
> interceptors run in the sandboxed workflow environment, this functionality has been extracted out to the worker.

```go
opts := worker.Options{
NexusEndpointResolver: func(ctx workflow.Context, service string) temporalnexus.Endpoint {
return temporalnexus.NewNamedEndpoint("prod-" + service)
}
}
```

### Interceptors

For now we'll only intercept outbound APIs from a workflow, extending the `WorkflowOutboundInterceptor` interface.
More interceptors are likely to come later.

```go
type WorkflowOutboundInterceptor interface {
ScheduleNexusOperation(ctx Context, service, operation string, input any, options NexusOperationOptions) NexusOperationFuture
NewNexusClient(service, endpoint string) workflow.NexusClient
RequestCancelNexusOperation(ctx Context, service, operation, id string, options nexus.CancelOperationOptions) error
}
```
Expand Down