mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
e8d2d2a519
## What changed? - Fix superficial code consistency and structure issues - Rename invokable -> invocable - Rename `chasmInvokation` to `invocationInternal` - it is used for internal, cross-shard requests - Rename `nexusInvokation` to `invocationOutbound` - it is used for outbound, cross-namespace or cross-cell requests - Add some new doc strings ## Why? - Make code a bit easier to reason about --------- Co-authored-by: Sam <[email protected]>
111 lines
3.4 KiB
Go
111 lines
3.4 KiB
Go
package callback
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/http/httptrace"
|
|
"time"
|
|
|
|
"github.com/nexus-rpc/sdk-go/nexus"
|
|
"go.temporal.io/server/chasm"
|
|
callbackspb "go.temporal.io/server/chasm/lib/callback/gen/callbackpb/v1"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/namespace"
|
|
commonnexus "go.temporal.io/server/common/nexus"
|
|
"go.temporal.io/server/common/nexus/nexusrpc"
|
|
queuescommon "go.temporal.io/server/service/history/queues/common"
|
|
queueserrors "go.temporal.io/server/service/history/queues/errors"
|
|
)
|
|
|
|
// invocableOutbound is an invocable that delivers the Nexus operation completion data to an external destination for
|
|
// cross-namespace or cross-cell callbacks.
|
|
type invocableOutbound struct {
|
|
callback *callbackspb.Callback_Nexus
|
|
completion nexusrpc.CompleteOperationOptions
|
|
workflowID, runID string
|
|
attempt int32
|
|
}
|
|
|
|
func (n invocableOutbound) WrapError(result invocationResult, err error) error {
|
|
if retry, ok := result.(invocationResultRetry); ok {
|
|
return queueserrors.NewDestinationDownError(retry.err.Error(), err)
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (n invocableOutbound) Invoke(
|
|
ctx context.Context,
|
|
ns *namespace.Namespace,
|
|
h *invocationTaskHandler,
|
|
task *callbackspb.InvocationTask,
|
|
taskAttr chasm.TaskAttributes,
|
|
) invocationResult {
|
|
if h.httpTraceProvider != nil {
|
|
traceLogger := log.With(h.logger,
|
|
tag.WorkflowNamespace(ns.Name().String()),
|
|
tag.Operation("CompleteNexusOperation"),
|
|
tag.String("destination", taskAttr.Destination),
|
|
tag.WorkflowID(n.workflowID),
|
|
tag.WorkflowRunID(n.runID),
|
|
tag.AttemptStart(time.Now().UTC()),
|
|
tag.Attempt(n.attempt),
|
|
)
|
|
if trace := h.httpTraceProvider.NewTrace(n.attempt, traceLogger); trace != nil {
|
|
ctx = httptrace.WithClientTrace(ctx, trace)
|
|
}
|
|
}
|
|
|
|
client := nexusrpc.NewCompletionHTTPClient(nexusrpc.CompletionHTTPClientOptions{
|
|
HTTPCaller: h.httpCallerProvider(queuescommon.NamespaceIDAndDestination{
|
|
NamespaceID: ns.ID().String(),
|
|
Destination: taskAttr.Destination,
|
|
}),
|
|
Serializer: commonnexus.PayloadSerializer,
|
|
})
|
|
// Make the call and record metrics.
|
|
startTime := time.Now()
|
|
|
|
n.completion.Header = n.callback.Header
|
|
err := client.CompleteOperation(ctx, n.callback.Url, n.completion)
|
|
|
|
namespaceTag := metrics.NamespaceTag(ns.Name().String())
|
|
destTag := metrics.DestinationTag(taskAttr.Destination)
|
|
outcomeTag := metrics.OutcomeTag(outcomeTag(ctx, err))
|
|
h.metricsHandler.Counter(RequestCounter.Name()).Record(1, namespaceTag, destTag, outcomeTag)
|
|
h.metricsHandler.Timer(RequestLatencyHistogram.Name()).Record(time.Since(startTime), namespaceTag, destTag, outcomeTag)
|
|
|
|
if err != nil {
|
|
retryable := isRetryableCallError(err)
|
|
h.logger.Error("Callback request failed", tag.Error(err), tag.Bool("retryable", retryable))
|
|
if retryable {
|
|
return invocationResultRetry{err}
|
|
}
|
|
return invocationResultFail{err}
|
|
}
|
|
return invocationResultOK{}
|
|
}
|
|
|
|
func isRetryableCallError(err error) bool {
|
|
var handlerError *nexus.HandlerError
|
|
if errors.As(err, &handlerError) {
|
|
return handlerError.Retryable()
|
|
}
|
|
return true
|
|
}
|
|
|
|
func outcomeTag(callCtx context.Context, callErr error) string {
|
|
if callErr != nil {
|
|
if callCtx.Err() != nil {
|
|
return "request-timeout"
|
|
}
|
|
var handlerErr *nexus.HandlerError
|
|
if errors.As(callErr, &handlerErr) {
|
|
return "handler-error:" + string(handlerErr.Type)
|
|
}
|
|
return "unknown-error"
|
|
}
|
|
return "success"
|
|
}
|