avoid duplicate egress (#701)

pkg/errors/errors.go

@@ -99,6 +99,7 @@ var (
 	ErrNoCompatibleCodec          = psrpc.NewErrorf(psrpc.InvalidArgument, "no supported codec is compatible with all outputs")
 	ErrNoCompatibleFileOutputType = psrpc.NewErrorf(psrpc.InvalidArgument, "no supported file output type is compatible with the selected codecs")
 	ErrEgressNotFound             = psrpc.NewErrorf(psrpc.NotFound, "egress not found")
+	ErrEgressAlreadyExists        = psrpc.NewErrorf(psrpc.AlreadyExists, "egress already exists")
 	ErrSubscriptionFailed         = psrpc.NewErrorf(psrpc.Unavailable, "failed to subscribe to track")
 	ErrNotEnoughCPU               = psrpc.NewErrorf(psrpc.Unavailable, "not enough CPU")
 	ErrShuttingDown               = psrpc.NewErrorf(psrpc.Unavailable, "server is shutting down")

pkg/server/server_rpc.go

@@ -46,7 +46,10 @@ func (s *Server) StartEgress(ctx context.Context, req *rpc.StartEgressRequest) (
 		s.activeRequests.Dec()
 		return nil, errors.ErrShuttingDown
 	}
-
+	if s.AlreadyExists(req.EgressId) {
+		s.activeRequests.Dec()
+		return nil, errors.ErrEgressAlreadyExists
+	}
 	if err := s.monitor.AcceptRequest(req); err != nil {
 		s.activeRequests.Dec()
 		return nil, err

pkg/service/process.go

@@ -101,6 +101,14 @@ func (pm *ProcessManager) GetContext(egressID string) context.Context {
 	return context.Background()
 }
 
+func (pm *ProcessManager) AlreadyExists(egressID string) bool {
+	pm.mu.RLock()
+	defer pm.mu.RUnlock()
+
+	_, ok := pm.activeHandlers[egressID]
+	return ok
+}
+
 func (pm *ProcessManager) HandlerStarted(egressID string) error {
 	pm.mu.RLock()
 	defer pm.mu.RUnlock()

pkg/stats/monitor.go

@@ -189,6 +189,9 @@ func (m *Monitor) AcceptRequest(req *rpc.StartEgressRequest) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
+	if m.pending[req.EgressId] != nil {
+		return errors.ErrEgressAlreadyExists
+	}
 	if !m.canAcceptRequestLocked(req) {
 		return errors.ErrNotEnoughCPU
 	}