tests tests tests

Author: ad-astra-video

common/testutil.go

@@ -90,12 +90,23 @@ func IgnoreRoutines() []goleak.Option {
 		"github.com/livepeer/go-livepeer/server.(*LivepeerServer).HandlePush.func1", "github.com/rjeczalik/notify.(*nonrecursiveTree).dispatch",
 		"github.com/rjeczalik/notify.(*nonrecursiveTree).internal", "github.com/livepeer/lpms/stream.NewBasicRTMPVideoStream.func1", "github.com/patrickmn/go-cache.(*janitor).Run",
 		"github.com/golang/glog.(*fileSink).flushDaemon", "github.com/livepeer/go-livepeer/core.(*LivepeerNode).transcodeFrames.func2", "github.com/ipfs/go-log/writer.(*MirrorWriter).logRoutine",
-		"github.com/livepeer/go-livepeer/core.(*Balances).StartCleanup",
+		"github.com/livepeer/go-livepeer/core.(*Balances).StartCleanup", "github.com/livepeer/go-livepeer/server.startTrickleSubscribe.func2", "github.com/livepeer/go-livepeer/server.startTrickleSubscribe",
+		"net/http.(*persistConn).writeLoop", "net/http.(*persistConn).readLoop", "io.(*pipe).read",
+		"github.com/livepeer/go-livepeer/media.gatherIncomingTracks",
 	}
 
-	res := make([]goleak.Option, 0, len(funcs2ignore))
+	// Functions that might have other functions on top of their stack (like time.Sleep)
+	// These need to be ignored with IgnoreAnyFunction instead of IgnoreTopFunction
+	funcsAnyIgnore := []string{
+		"github.com/livepeer/go-livepeer/server.ffmpegOutput",
+	}
+
+	res := make([]goleak.Option, 0, len(funcs2ignore)+len(funcsAnyIgnore))
 	for _, f := range funcs2ignore {
 		res = append(res, goleak.IgnoreTopFunction(f))
 	}
+	for _, f := range funcsAnyIgnore {
+		res = append(res, goleak.IgnoreAnyFunction(f))
+	}
 	return res
 }

core/livepeernode.go

@@ -195,15 +195,23 @@ func (n *LivepeerNode) NewLivePipeline(requestID, streamID, pipeline string, str
 	n.LiveMu.Lock()
 	defer n.LiveMu.Unlock()
 	n.LivePipelines[streamID] = &LivePipeline{
-		RequestID:    requestID,
-		Pipeline:     pipeline,
-		StreamCtx:    streamCtx,
-		streamParams: streamParams,
-		streamCancel: streamCancel,
+		RequestID:     requestID,
+		StreamID:      streamID,
+		Pipeline:      pipeline,
+		StreamCtx:     streamCtx,
+		streamParams:  streamParams,
+		streamCancel:  streamCancel,
+		streamRequest: streamRequest,
 	}
 	return n.LivePipelines[streamID]
 }
 
+func (n *LivepeerNode) RemoveLivePipeline(streamID string) {
+	n.LiveMu.Lock()
+	defer n.LiveMu.Unlock()
+	delete(n.LivePipelines, streamID)
+}
+
 func (p *LivePipeline) StreamParams() interface{} {
 	return p.streamParams
 }
@@ -217,7 +225,10 @@ func (p *LivePipeline) StreamRequest() []byte {
 }
 
 func (p *LivePipeline) StopStream(err error) {
-	p.StopControl()
+	if p.StopControl != nil {
+		p.StopControl()
+	}
+
 	p.streamCancel(err)
 }
 

server/job_rpc.go

@@ -864,10 +864,22 @@ func processPayment(ctx context.Context, orch Orchestrator, sender ethcommon.Add
 }
 
 func createPayment(ctx context.Context, jobReq *JobRequest, orchToken *core.JobToken, node *core.LivepeerNode) (string, error) {
+	if orchToken == nil {
+		return "", errors.New("orchestrator token is nil, cannot create payment")
+	}
+	//if no sender or ticket params, no payment
+	if node.Sender == nil {
+		return "", errors.New("no ticket sender available, cannot create payment")
+	}
+	if orchToken.TicketParams == nil {
+		return "", errors.New("no ticket params available, cannot create payment")
+	}
+
 	var payment *net.Payment
 	createTickets := true
 	clog.Infof(ctx, "creating payment for job request %s", jobReq.Capability)
 	sender := ethcommon.HexToAddress(jobReq.Sender)
+
 	orchAddr := ethcommon.BytesToAddress(orchToken.TicketParams.Recipient)
 	sessionID := node.Sender.StartSession(*pmTicketParams(orchToken.TicketParams))