Skip to content

Consolidate kitchen sink Nexus operations #361

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

Merged
stephanos merged 5 commits into from
Sep 9, 2026
Merged

Consolidate kitchen sink Nexus operations #361

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
972e4ee
Consolidate kitchen sink Nexus operations
stephanos Sep 7, 2026
660da9a
Address Nexus operation review feedback
stephanos Sep 7, 2026
06fe58a
Rename kitchen sink Nexus operation descriptor
stephanos Sep 8, 2026
eab86a9
Update client_action_executor.go
stephanos Sep 8, 2026
eea2875
Update kitchen_sink.go
stephanos Sep 9, 2026
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
12 changes: 10 additions & 2 deletions clioptions/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,13 @@ import (
"fmt"
"os"

"github.com/golang/protobuf/proto"
"github.com/spf13/pflag"
"github.com/temporalio/omes/metrics"
"go.temporal.io/api/common/v1"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/converter"
"go.uber.org/zap"
"google.golang.org/protobuf/proto"
)

const AUTH_HEADER_ENV_VAR = "TEMPORAL_OMES_AUTH_HEADER"
Expand Down Expand Up @@ -176,7 +176,15 @@ func (p *PassThroughPayloadConverter) FromPayload(payload *common.Payload, value
if err != nil {
return fmt.Errorf("unable to decode raw payload: %w", err)
}
return converter.GetDefaultDataConverter().FromPayload(innerPayload, valuePtr)
switch target := valuePtr.(type) {
case *common.Payload:
// The caller wants the encoded Payload itself, not its decoded value.
proto.Reset(target)
proto.Merge(target, innerPayload)
return nil
default:
return converter.GetDefaultDataConverter().FromPayload(innerPayload, target)
}
}

func (p *PassThroughPayloadConverter) ToString(payload *common.Payload) string {
Expand Down
132 changes: 59 additions & 73 deletions loadgen/kitchen-sink-gen/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,14 @@ use crate::protos::temporal::{
do_signal::DoSignalActions,
do_update, execute_activity_action,
execute_activity_action::{ClientActivity, PayloadActivity},
with_start_client_action, Action, ActionSet, AwaitWorkflowState, AwaitableChoice,
ClientAction, ClientActionSet, ClientSequence, DoQuery, DoSignal, DoUpdate,
ExecuteActivityAction, ExecuteChildWorkflowAction, ExecuteNexusOperation,
HandlerInvocation, RemoteActivityOptions, ReturnResultAction, SetPatchMarkerAction,
TestInput, TimerAction, UpsertMemoAction, UpsertSearchAttributesAction,
WithStartClientAction, WorkflowInput, WorkflowState,
nexus_operation_request, nexus_workflow_action, with_start_client_action, Action,
ActionSet, AwaitWorkflowState, AwaitableChoice, ClientAction, ClientActionSet,
ClientSequence, DoQuery, DoSignal, DoUpdate, ExecuteActivityAction,
ExecuteChildWorkflowAction, ExecuteNexusOperation, HandlerInvocation,
NexusOperationRequest, NexusWorkflowAction, NexusWorkflowStartOptions,
RemoteActivityOptions, ReturnResultAction, SetPatchMarkerAction, TestInput, TimerAction,
UpsertMemoAction, UpsertSearchAttributesAction, WithStartClientAction, WorkflowInput,
WorkflowState,
},
};
use anyhow::Error;
Expand Down Expand Up @@ -618,12 +620,7 @@ impl<'a> Arbitrary<'a> for Action {
} else if chances.nested_action_set(action_kind) {
action::Variant::NestedActionSet(u.arbitrary()?)
} else if chances.nexus_operation(action_kind) {
if ARB_CONTEXT.with_borrow(|c| c.action_set_nest_level >= 1) {
// Nested nexus operations are not supported, use echo-sync instead
action::Variant::NexusOperation(ExecuteNexusOperation::echo_sync(u)?)
} else {
action::Variant::NexusOperation(u.arbitrary()?)
}
action::Variant::NexusOperation(u.arbitrary()?)
} else {
unreachable!()
};
Expand Down Expand Up @@ -712,70 +709,48 @@ impl<'a> Arbitrary<'a> for ExecuteChildWorkflowAction {
}
}

static NEXUS_OPERATIONS: [&str; 2] = ["echo-sync", "echo-async"];

impl ExecuteNexusOperation {
fn echo_sync(u: &mut Unstructured<'_>) -> arbitrary::Result<Self> {
let val = format!("nexus-test-{}", u.int_in_range(1..=1000)?);
Ok(Self {
endpoint: ARB_CONTEXT.with_borrow(|c| c.nexus_endpoint.clone()),
operation: "echo-sync".to_string(),
input: val.clone(),
expected_output: val,
// echo-sync completes immediately, so only WaitFinish is valid.
awaitable_choice: Some(AwaitableChoice {
condition: Some(awaitable_choice::Condition::WaitFinish(())),
}),
before_actions: vec![],
handler_workflow_id: String::new(),
handler_workflow_id_conflict_policy: 0,
wait_for_signal: false,
})
}
}

impl<'a> Arbitrary<'a> for ExecuteNexusOperation {
fn arbitrary(u: &mut Unstructured<'a>) -> arbitrary::Result<Self> {
let &operation = u.choose(&NEXUS_OPERATIONS)?;

if operation == "echo-sync" {
return Self::echo_sync(u);
}

let endpoint = ARB_CONTEXT.with_borrow(|c| c.nexus_endpoint.clone());
let val = format!("nexus-test-{}", u.int_in_range(1..=1000)?);
let (input, expected_output) = (val.clone(), val);

// Randomly generate before_actions for echo-async operations
let before_actions = if u.ratio(1, 3)? {
ARB_CONTEXT.with_borrow_mut(|c| c.action_set_nest_level += 1);
let num_actions =
u.int_in_range(1..=ARB_CONTEXT.with_borrow(|c| c.config.max_actions_per_set))?;
let mut actions: Vec<Action> = Vec::with_capacity(num_actions);
for _ in 0..num_actions {
actions.push(u.arbitrary()?);
}
ARB_CONTEXT.with_borrow_mut(|c| c.action_set_nest_level -= 1);
vec![ActionSet {
actions,
concurrent: false,
}]
if u.ratio(1, 2)? {
let val = format!("nexus-test-{}", u.int_in_range(1..=1000)?);
Ok(Self {
endpoint: ARB_CONTEXT.with_borrow(|c| c.nexus_endpoint.clone()),
operation: "execute".to_string(),
expected_output: Some(json_payload(&val)),
// Echo completes immediately, so only WaitFinish is valid.
awaitable_choice: Some(AwaitableChoice {
condition: Some(awaitable_choice::Condition::WaitFinish(())),
}),
input: Some(NexusOperationRequest {
action: Some(nexus_operation_request::Action::Echo(val)),
}),
})
} else {
vec![]
};

// echo-async supports all awaitable choices including cancellation.
Ok(Self {
endpoint,
operation: operation.to_string(),
input,
awaitable_choice: Some(u.arbitrary()?),
expected_output,
before_actions,
handler_workflow_id: String::new(),
handler_workflow_id_conflict_policy: 0,
wait_for_signal: false,
})
Ok(Self {
endpoint: ARB_CONTEXT.with_borrow(|c| c.nexus_endpoint.clone()),
operation: "execute".to_string(),
expected_output: None,
awaitable_choice: Some(u.arbitrary()?),
input: Some(NexusOperationRequest {
action: Some(nexus_operation_request::Action::WorkflowAction(
NexusWorkflowAction {
start_options: Some(NexusWorkflowStartOptions {
workflow_input: Some(WorkflowInput {
initial_actions: vec![mk_action_set([ReturnResultAction {
return_this: Some(empty_payload()),
}
.into()])],
..Default::default()
}),
..Default::default()
}),
action: Some(nexus_workflow_action::Action::Start(())),
..Default::default()
},
)),
}),
})
}
}
}

Expand Down Expand Up @@ -1036,3 +1011,14 @@ fn empty_payload() -> Payload {
data: vec![], // Empty
}
}

fn json_payload(value: &str) -> Payload {
Payload {
metadata: {
let mut m = HashMap::new();
m.insert("encoding".to_string(), "json/plain".into());
m
},
data: serde_json::to_vec(value).expect("serializes"),
}
}
120 changes: 96 additions & 24 deletions loadgen/kitchen_sink_executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -944,8 +944,38 @@ func TestKitchenSink(t *testing.T) {
&Action{
Variant: &Action_NexusOperation{
NexusOperation: &ExecuteNexusOperation{
Operation: "echo-sync",
Input: "hello",
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_Echo{Echo: "hello"},
},
AwaitableChoice: &AwaitableChoice{
Condition: &AwaitableChoice_WaitFinish{
WaitFinish: &emptypb.Empty{},
},
},
},
},
}),
},
},
historyMatcher: PartialHistoryMatcher(`
NexusOperationScheduled {"operation":"execute"}
NexusOperationCompleted`),
expectedUnsupportedErrs: nexusUnsupportedSDKs,
},
{
name: "NexusOperation/Sync/ExpectedOutputMismatch",

@stephanos stephanos Sep 7, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Only net new test here.

testInput: &TestInput{
WorkflowInput: &WorkflowInput{
InitialActions: ListActionSet(
&Action{
Variant: &Action_NexusOperation{
NexusOperation: &ExecuteNexusOperation{
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_Echo{Echo: "hello"},
},
ExpectedOutput: ConvertToPayload("goodbye"),
AwaitableChoice: &AwaitableChoice{
Condition: &AwaitableChoice_WaitFinish{
WaitFinish: &emptypb.Empty{},
Expand All @@ -957,9 +987,10 @@ func TestKitchenSink(t *testing.T) {
},
},
historyMatcher: PartialHistoryMatcher(`
NexusOperationScheduled {"operation":"echo-sync"}
NexusOperationScheduled {"operation":"execute"}
NexusOperationCompleted`),
expectedUnsupportedErrs: nexusUnsupportedSDKs,
expectedWorkflowError: `goodbye`,
},
{
name: "NexusOperation/Async",
Expand All @@ -969,11 +1000,22 @@ func TestKitchenSink(t *testing.T) {
&Action{
Variant: &Action_NexusOperation{
NexusOperation: &ExecuteNexusOperation{
Operation: "echo-async",
Input: "world",
BeforeActions: ListActionSet(
NewTimerAction(1),
),
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_WorkflowAction{
WorkflowAction: &NexusWorkflowAction{
StartOptions: &NexusWorkflowStartOptions{
WorkflowInput: &WorkflowInput{
InitialActions: ListActionSet(
NewTimerAction(1),
NewEmptyReturnResultAction(),
),
},
},
Action: &NexusWorkflowAction_Start{Start: &emptypb.Empty{}},
},
},
},
AwaitableChoice: &AwaitableChoice{
Condition: &AwaitableChoice_WaitFinish{
WaitFinish: &emptypb.Empty{},
Expand All @@ -985,7 +1027,7 @@ func TestKitchenSink(t *testing.T) {
},
},
historyMatcher: PartialHistoryMatcher(`
NexusOperationScheduled {"operation":"echo-async"}
NexusOperationScheduled {"operation":"execute"}
NexusOperationStarted
NexusOperationCompleted`),
expectedUnsupportedErrs: nexusUnsupportedSDKs,
Expand All @@ -998,10 +1040,21 @@ func TestKitchenSink(t *testing.T) {
&Action{
Variant: &Action_NexusOperation{
NexusOperation: &ExecuteNexusOperation{
Operation: "echo-async",
BeforeActions: ListActionSet(
NewAwaitWorkflowStateAction("never", "resolves"),
),
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_WorkflowAction{
WorkflowAction: &NexusWorkflowAction{
StartOptions: &NexusWorkflowStartOptions{
WorkflowInput: &WorkflowInput{
InitialActions: ListActionSet(
NewAwaitWorkflowStateAction("never", "resolves"),
),
},
},
Action: &NexusWorkflowAction_Start{Start: &emptypb.Empty{}},
},
},
},
AwaitableChoice: &AwaitableChoice{
Condition: &AwaitableChoice_CancelAfterStarted{
CancelAfterStarted: &emptypb.Empty{},
Expand All @@ -1013,7 +1066,7 @@ func TestKitchenSink(t *testing.T) {
},
},
historyMatcher: PartialHistoryMatcher(`
NexusOperationScheduled {"operation":"echo-async"}
NexusOperationScheduled {"operation":"execute"}
NexusOperationStarted
NexusOperationCancelRequested
NexusOperationCanceled`),
Expand All @@ -1027,8 +1080,10 @@ func TestKitchenSink(t *testing.T) {
&Action{
Variant: &Action_NexusOperation{
NexusOperation: &ExecuteNexusOperation{
Operation: "echo-sync",
Input: "abandoned",
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_Echo{Echo: "abandoned"},
},
AwaitableChoice: &AwaitableChoice{
Condition: &AwaitableChoice_Abandon{
Abandon: &emptypb.Empty{},
Expand All @@ -1040,7 +1095,7 @@ func TestKitchenSink(t *testing.T) {
},
},
historyMatcher: PartialHistoryMatcher(`
NexusOperationScheduled {"operation":"echo-sync"}
NexusOperationScheduled {"operation":"execute"}
...
WorkflowExecutionCompleted`),
expectedUnsupportedErrs: nexusUnsupportedSDKs,
Expand All @@ -1054,9 +1109,22 @@ func TestKitchenSink(t *testing.T) {
ClientActions(&ClientAction{
Variant: &ClientAction_DoStandaloneNexusOperation{
DoStandaloneNexusOperation: &DoStandaloneNexusOperation{
// Endpoint filled by PrepareTestInput
Service: "kitchen-sink",
Operation: "echo-async",
Operation: &ExecuteNexusOperation{
// Endpoint filled by PrepareTestInput
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_WorkflowAction{
WorkflowAction: &NexusWorkflowAction{
StartOptions: &NexusWorkflowStartOptions{
WorkflowInput: &WorkflowInput{
InitialActions: ListActionSet(NewEmptyReturnResultAction()),
},
},
Action: &NexusWorkflowAction_Start{Start: &emptypb.Empty{}},
},
},
},
},
},
},
}),
Expand All @@ -1080,9 +1148,13 @@ func TestKitchenSink(t *testing.T) {
ClientActions(&ClientAction{
Variant: &ClientAction_DoStandaloneNexusOperation{
DoStandaloneNexusOperation: &DoStandaloneNexusOperation{
// Endpoint filled by PrepareTestInput
Service: "kitchen-sink",
Operation: "echo-sync",
Operation: &ExecuteNexusOperation{
// Endpoint filled by PrepareTestInput
Operation: KitchenSinkNexusOperationName,
Input: &NexusOperationRequest{
Action: &NexusOperationRequest_Echo{Echo: "hello"},
},
},
},
},
}),
Expand Down Expand Up @@ -1256,7 +1328,7 @@ func testForSDK(
if clientSeq := action.GetExecActivity().GetClient().GetClientSequence(); clientSeq != nil {
for _, cas := range clientSeq.ActionSets {
for _, ca := range cas.Actions {
if sno := ca.GetDoStandaloneNexusOperation(); sno != nil && sno.Endpoint == "" {
if sno := ca.GetDoStandaloneNexusOperation().GetOperation(); sno != nil && sno.Endpoint == "" {
sno.Endpoint = nexusEndpoint
}
if sa := ca.GetDoStandaloneActivity(); sa.GetActivity() != nil && sa.GetActivity().TaskQueue == "" {
Expand Down
Loading
Loading