Handle child node timeout in branch nodes (#6678)

Author: Sovietaced

flytepropeller/pkg/controller/nodes/branch/handler.go

@@ -172,7 +172,7 @@ func (b *branchHandler) recurseDownstream(ctx context.Context, nCtx interfaces.N
 		return handler.DoTransition(handler.TransitionTypeEphemeral, phase), nil
 	}
 
-	if downstreamStatus.HasFailed() {
+	if downstreamStatus.HasFailed() || downstreamStatus.HasTimedOut() {
 		return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoFailureErr(downstreamStatus.Err, nil)), nil
 	}
 

flytepropeller/pkg/controller/nodes/branch/handler_test.go

@@ -181,6 +181,8 @@ func TestBranchHandler_RecurseDownstream(t *testing.T) {
 			bn, false, handler.EPhaseRunning, v1alpha1.NodePhaseRunning, true, ""},
 		{"childFailure", interfaces.NodeStatusFailed(expectedError), nil,
 			bn, false, handler.EPhaseFailed, v1alpha1.NodePhaseFailed, true, ""},
+		{"childTimedOut", interfaces.NodeStatusTimedOut, nil,
+			bn, false, handler.EPhaseFailed, v1alpha1.NodePhaseTimedOut, true, ""},
 		{"childComplete", interfaces.NodeStatusComplete, nil,
 			bn, false, handler.EPhaseSuccess, v1alpha1.NodePhaseSucceeded, true, ""},
 		{"childCompleteNoOutputs", interfaces.NodeStatusComplete, nil,