mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
31f4ee5d4c
## What Dispatches worker commands (starting with activity cancellation) to workers via their Nexus control queue. When the outbound queue processes a `WorkerCommandsTask`, the dispatcher sends an `ExecuteCommands` Nexus operation to the worker's control queue via `DispatchNexusTask`. - Retries are capped at 3 attempts since these commands are best-effort (the activity will eventually time out anyway). Uses the retryable matching client so transient RPC errors (ShardOwnershipLost, Unavailable) are retried at the RPC layer without consuming queue-level attempts. - Stores the clock used to generate the task token in the ActivityInfo. This is needed to reconstruct the same task token when dispatching to the worker. Otherwise, it will not match the token expected by the sdk. Suggested review order: `worker_commands_task_dispatcher.go` → `nexus_dispatch_response.go` → `recordactivitytaskstarted/api.go`. ## Why To support activity cancellation without activity heartbeat. This is the dispatch leg of the flow: 1. [#9231] Store `worker_control_task_queue` in `ActivityInfo` at activity start. 2. [#9232] On `RequestCancelActivityTask`, batch commands by control queue into `WorkerCommandsTask` outbound tasks. 3. **[This PR]** Dispatch each task as a Nexus `ExecuteCommands` operation to the worker, with a 3-attempt retry cap. 4. [SDK] Worker receives the cancel command and cancels the running activity. Gated by dynamic config `EnableCancelActivityWorkerCommand` (default: off). ## How did you test it? - **Unit tests** cover all dispatch outcomes (success, RPC error, timeout, worker error, feature-flag-off, max-attempts-exceeded) and response-to-error conversion paths. - **Functional test** verifies end-to-end: cancel request → Nexus dispatch → correct payload arrives on the control queue, and asserts that the cancel command's task token matches the one from the original activity poll response. --------- Co-authored-by: rkannan82 <[email protected]> Co-authored-by: Hanzo Dev <[email protected]>
161 lines
4.7 KiB
Go
161 lines
4.7 KiB
Go
package history
|
|
|
|
import (
|
|
"testing"
|
|
|
|
"github.com/nexus-rpc/sdk-go/nexus"
|
|
"github.com/stretchr/testify/require"
|
|
failurepb "go.temporal.io/api/failure/v1"
|
|
nexuspb "go.temporal.io/api/nexus/v1"
|
|
"go.temporal.io/sdk/temporal"
|
|
"go.temporal.io/server/api/matchingservice/v1"
|
|
)
|
|
|
|
func TestDispatchResponseToError_SyncSuccess(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{
|
|
Outcome: &matchingservice.DispatchNexusTaskResponse_Response{
|
|
Response: &nexuspb.Response{
|
|
Variant: &nexuspb.Response_StartOperation{
|
|
StartOperation: &nexuspb.StartOperationResponse{
|
|
Variant: &nexuspb.StartOperationResponse_SyncSuccess{
|
|
SyncSuccess: &nexuspb.StartOperationResponse_Sync{},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
err := dispatchResponseToError(resp)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func TestDispatchResponseToError_AsyncSuccess(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{
|
|
Outcome: &matchingservice.DispatchNexusTaskResponse_Response{
|
|
Response: &nexuspb.Response{
|
|
Variant: &nexuspb.Response_StartOperation{
|
|
StartOperation: &nexuspb.StartOperationResponse{
|
|
Variant: &nexuspb.StartOperationResponse_AsyncSuccess{
|
|
AsyncSuccess: &nexuspb.StartOperationResponse_Async{
|
|
OperationId: "test-op-id",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
err := dispatchResponseToError(resp)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func TestDispatchResponseToError_RequestTimeout(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{
|
|
Outcome: &matchingservice.DispatchNexusTaskResponse_RequestTimeout{
|
|
RequestTimeout: &matchingservice.DispatchNexusTaskResponse_Timeout{},
|
|
},
|
|
}
|
|
err := dispatchResponseToError(resp)
|
|
require.Error(t, err)
|
|
|
|
var handlerErr *nexus.HandlerError
|
|
require.ErrorAs(t, err, &handlerErr)
|
|
require.Equal(t, nexus.HandlerErrorTypeUpstreamTimeout, handlerErr.Type)
|
|
}
|
|
|
|
func TestDispatchResponseToError_WorkerFailure(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{
|
|
Outcome: &matchingservice.DispatchNexusTaskResponse_Failure{
|
|
Failure: &failurepb.Failure{
|
|
Message: "bad request from worker",
|
|
FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
|
|
ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
|
|
Type: "SomeError",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
err := dispatchResponseToError(resp)
|
|
require.Error(t, err)
|
|
|
|
var appErr *temporal.ApplicationError
|
|
require.ErrorAs(t, err, &appErr)
|
|
require.Equal(t, "SomeError", appErr.Type())
|
|
}
|
|
|
|
func TestDispatchResponseToError_OperationFailure_ApplicationError(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{
|
|
Outcome: &matchingservice.DispatchNexusTaskResponse_Response{
|
|
Response: &nexuspb.Response{
|
|
Variant: &nexuspb.Response_StartOperation{
|
|
StartOperation: &nexuspb.StartOperationResponse{
|
|
Variant: &nexuspb.StartOperationResponse_Failure{
|
|
Failure: &failurepb.Failure{
|
|
Message: "activity failed",
|
|
FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
|
|
ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
|
|
Type: "SomeError",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
err := dispatchResponseToError(resp)
|
|
require.Error(t, err)
|
|
|
|
var appErr *temporal.ApplicationError
|
|
require.ErrorAs(t, err, &appErr)
|
|
require.Equal(t, "SomeError", appErr.Type())
|
|
}
|
|
|
|
func TestDispatchResponseToError_OperationFailure_CanceledError(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{
|
|
Outcome: &matchingservice.DispatchNexusTaskResponse_Response{
|
|
Response: &nexuspb.Response{
|
|
Variant: &nexuspb.Response_StartOperation{
|
|
StartOperation: &nexuspb.StartOperationResponse{
|
|
Variant: &nexuspb.StartOperationResponse_Failure{
|
|
Failure: &failurepb.Failure{
|
|
Message: "canceled",
|
|
FailureInfo: &failurepb.Failure_CanceledFailureInfo{
|
|
CanceledFailureInfo: &failurepb.CanceledFailureInfo{},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
err := dispatchResponseToError(resp)
|
|
require.Error(t, err)
|
|
|
|
var cancelErr *temporal.CanceledError
|
|
require.ErrorAs(t, err, &cancelErr)
|
|
}
|
|
|
|
func TestDispatchResponseToError_EmptyOutcome(t *testing.T) {
|
|
resp := &matchingservice.DispatchNexusTaskResponse{}
|
|
err := dispatchResponseToError(resp)
|
|
require.Error(t, err)
|
|
|
|
var handlerErr *nexus.HandlerError
|
|
require.ErrorAs(t, err, &handlerErr)
|
|
require.Equal(t, nexus.HandlerErrorTypeInternal, handlerErr.Type)
|
|
}
|
|
|
|
func TestStartOperationResponseToError_EmptyVariant(t *testing.T) {
|
|
resp := &nexuspb.StartOperationResponse{}
|
|
err := startOperationResponseToError(resp)
|
|
require.Error(t, err)
|
|
|
|
var handlerErr *nexus.HandlerError
|
|
require.ErrorAs(t, err, &handlerErr)
|
|
require.Equal(t, nexus.HandlerErrorTypeInternal, handlerErr.Type)
|
|
}
|