Do not enforce payload limits for system nexus endpoint (#9344)

Rely on the gRPC 4MB limit to enforce an upper bound. Operations may
have embedded payloads that would be validated on a per operation basis.

Author: bergundy

components/nexusoperations/workflow/commands.go

@@ -148,7 +148,7 @@ func (ch *commandHandler) HandleScheduleCommand(
 		}
 	}
 
-	if !validator.IsValidPayloadSize(attrs.Input.Size()) {
+	if attrs.Endpoint != commonnexus.SystemEndpoint && !validator.IsValidPayloadSize(attrs.Input.Size()) {
 		return workflow.FailWorkflowTaskError{
 			Cause:             enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_NEXUS_OPERATION_ATTRIBUTES,
 			Message:           "ScheduleNexusOperationCommandAttributes.Input exceeds size limit",

components/nexusoperations/workflow/commands_test.go

@@ -275,6 +275,29 @@ func TestHandleScheduleCommand(t *testing.T) {
 		require.Equal(t, 0, len(tcx.history.Events))
 	})
 
+	t.Run("system endpoint skips payload size validation", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		err := tcx.scheduleHandler(context.Background(), tcx.ms, commandValidator{maxPayloadSize: 1}, 1, &commandpb.Command{
+			Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
+				ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
+					Endpoint:  commonnexus.SystemEndpoint,
+					Service:   "service",
+					Operation: "op",
+					Input: &commonpb.Payload{
+						Data: []byte("ab"),
+					},
+				},
+			},
+		})
+		// Should NOT get payload size error; instead gets ProcessInput validation error (service not found).
+		var failWFTErr workflow.FailWorkflowTaskError
+		require.ErrorAs(t, err, &failWFTErr)
+		require.False(t, failWFTErr.TerminateWorkflow)
+		require.Equal(t, enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_NEXUS_OPERATION_ATTRIBUTES, failWFTErr.Cause)
+		require.NotContains(t, failWFTErr.Message, "Input exceeds size limit")
+		require.Empty(t, tcx.history.Events)
+	})
+
 	t.Run("exceeds max concurrent operations", func(t *testing.T) {
 		tcx := newTestContext(t, defaultConfig)
 		for range 2 {