Improve gRPC error handling, handle cancelling ffmpegReader

This commit is contained in:
Rob Watson 2021-11-08 14:56:25 +01:00
parent 06697dc1b1
commit 1496473620
3 changed files with 43 additions and 13 deletions

View File

@ -33,6 +33,13 @@ func newFfmpegReader(ctx context.Context, input io.Reader, arg ...string) (*ffmp
return &ffmpegReader{ReadCloser: r, cmd: cmd}, nil return &ffmpegReader{ReadCloser: r, cmd: cmd}, nil
} }
func (r *ffmpegReader) Cancel() error {
if err := r.cmd.Process.Kill(); err != nil {
return fmt.Errorf("error killing ffmpeg process: %v", err)
}
return nil
}
func (r *ffmpegReader) Close() error { func (r *ffmpegReader) Close() error {
state, err := r.cmd.Process.Wait() state, err := r.cmd.Process.Wait()

View File

@ -377,7 +377,7 @@ func (s *MediaSetService) getAudioFromYoutube(ctx context.Context, mediaSet stor
type getAudioFromYoutubeState struct { type getAudioFromYoutubeState struct {
*fetchAudioProgressReader *fetchAudioProgressReader
ffmpegReader io.ReadCloser ffmpegReader *ffmpegReader
uploader *multipartUploadWriter uploader *multipartUploadWriter
s3Bucket, s3Key string s3Bucket, s3Key string
store Store store Store
@ -404,6 +404,12 @@ outer:
} }
} }
if err != nil {
if cancelErr := s.ffmpegReader.Cancel(); cancelErr != nil {
log.Printf("error cancelling ffmpegreader: %v", cancelErr)
}
}
if readerErr := s.ffmpegReader.Close(); readerErr != nil { if readerErr := s.ffmpegReader.Close(); readerErr != nil {
if err == nil { if err == nil {
err = readerErr err = readerErr

View File

@ -17,25 +17,43 @@ import (
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
"google.golang.org/grpc/grpclog" "google.golang.org/grpc/grpclog"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/durationpb"
) )
type ResponseError struct { func init() {
error grpclog.SetLogger(log.New(os.Stdout, "server: ", log.LstdFlags))
}
ResponseCode codes.Code const (
// ts-proto generates code that automatically retries for a subset of gRPC
// response codes. To avoid invoking this behaviour, default to returning a
// Cancelled code. Later,
// See https://github.com/stephenh/ts-proto/blob/459b94f5b2988d58d186461332e888c3e511603a/src/generate-grpc-web.ts#L293
// and https://github.com/stephenh/ts-proto/pull/131.
defaultResponseCode = codes.Canceled
defaultResponseMessage = "An unexpected error occurred"
)
type ResponseError struct {
err error
s string
} }
func (r *ResponseError) Error() string { func (r *ResponseError) Error() string {
return fmt.Sprintf("An unexpected error occurred: %v (error code = %d).", r.error.Error(), r.ResponseCode) return fmt.Sprintf("unexpected error: %v", r.err.Error())
} }
func (r *ResponseError) Unwrap() error { func (r *ResponseError) Unwrap() error {
return r.error return r.err
} }
func newResponseError(err error, code codes.Code) *ResponseError { func (r *ResponseError) GRPCStatus() *status.Status {
return &ResponseError{error: err, ResponseCode: code} return status.New(defaultResponseCode, r.s)
}
func newResponseError(err error) *ResponseError {
return &ResponseError{err: err, s: defaultResponseMessage}
} }
type Options struct { type Options struct {
@ -61,7 +79,7 @@ type mediaSetServiceController struct {
func (c *mediaSetServiceController) Get(ctx context.Context, request *pbMediaSet.GetRequest) (*pbMediaSet.MediaSet, error) { func (c *mediaSetServiceController) Get(ctx context.Context, request *pbMediaSet.GetRequest) (*pbMediaSet.MediaSet, error) {
mediaSet, err := c.mediaSetService.Get(ctx, request.GetYoutubeId()) mediaSet, err := c.mediaSetService.Get(ctx, request.GetYoutubeId())
if err != nil { if err != nil {
return nil, newResponseError(err, codes.Unknown) return nil, newResponseError(err)
} }
result := pbMediaSet.MediaSet{ result := pbMediaSet.MediaSet{
@ -88,12 +106,12 @@ func (c *mediaSetServiceController) GetAudio(request *pbMediaSet.GetAudioRequest
id, err := uuid.Parse(request.GetId()) id, err := uuid.Parse(request.GetId())
if err != nil { if err != nil {
return newResponseError(err, codes.Unknown) return newResponseError(err)
} }
reader, err := c.mediaSetService.GetAudio(ctx, id, int(request.GetNumBins())) reader, err := c.mediaSetService.GetAudio(ctx, id, int(request.GetNumBins()))
if err != nil { if err != nil {
return newResponseError(err, codes.Unknown) return newResponseError(err)
} }
for { for {
@ -102,7 +120,7 @@ func (c *mediaSetServiceController) GetAudio(request *pbMediaSet.GetAudioRequest
if err == io.EOF { if err == io.EOF {
break break
} }
return newResponseError(err, codes.Unknown) return newResponseError(err)
} }
// TODO: consider using int32 throughout the backend flow to avoid this. // TODO: consider using int32 throughout the backend flow to avoid this.
@ -128,7 +146,6 @@ func Start(options Options) error {
fetchMediaSetService := media.NewMediaSetService(options.Store, options.YoutubeClient, options.S3Client) fetchMediaSetService := media.NewMediaSetService(options.Store, options.YoutubeClient, options.S3Client)
pbMediaSet.RegisterMediaSetServiceServer(grpcServer, &mediaSetServiceController{mediaSetService: fetchMediaSetService}) pbMediaSet.RegisterMediaSetServiceServer(grpcServer, &mediaSetServiceController{mediaSetService: fetchMediaSetService})
grpclog.SetLogger(log.New(os.Stdout, "server: ", log.LstdFlags))
// TODO: proper CORS support // TODO: proper CORS support
grpcWebServer := grpcweb.WrapServer(grpcServer, grpcweb.WithOriginFunc(func(string) bool { return true })) grpcWebServer := grpcweb.WrapServer(grpcServer, grpcweb.WithOriginFunc(func(string) bool { return true }))