Merge pull request #103 from wesen/bug/fix-error-handling-in-datatables

🚑 Fix error handling and also catch panics

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssm v1.37.5
 	github.com/go-go-golems/clay v0.1.14
 	github.com/go-go-golems/glazed v0.5.15
+	github.com/kucherenkovova/safegroup v1.0.2
 	github.com/labstack/echo/v4 v4.12.0
 	github.com/pkg/errors v0.9.1
 	github.com/rs/zerolog v1.30.0

go.sum

@@ -235,6 +235,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kucherenkovova/safegroup v1.0.2 h1:/FSf2zmnjhRdQUXFMuz1dlakxX9O8PhFW+N/rHzB/ZE=
+github.com/kucherenkovova/safegroup v1.0.2/go.mod h1:cTr0xuYzcIMvvmGHWgvYQZx5TJghNoYlNvWhJkJnc20=
 github.com/labstack/echo/v4 v4.12.0 h1:IKpw49IMryVB2p1a4dzwlhP1O2Tf2E0Ir/450lH+kI0=
 github.com/labstack/echo/v4 v4.12.0/go.mod h1:UP9Cr2DJXbOK3Kr9ONYzNowSh7HP0aG0ShAyycHSJvM=
 github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0=

pkg/glazed/handlers/datatables/datatables.go

@@ -18,10 +18,10 @@ import (
 	parka_middlewares "github.com/go-go-golems/parka/pkg/glazed/middlewares"
 	"github.com/go-go-golems/parka/pkg/render"
 	"github.com/go-go-golems/parka/pkg/render/layout"
+	"github.com/kucherenkovova/safegroup"
 	"github.com/labstack/echo/v4"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
-	"golang.org/x/sync/errgroup"
 	"html/template"
 	"io"
 	"time"
@@ -178,7 +178,7 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 	rowC := make(chan string, 100)
 
 	// buffered so that we don't hang on it when exiting
-	dt_.ErrorStream = make(chan string, 10)
+	dt_.ErrorStream = make(chan string, 1)
 
 	// columnsC is the channel where the column names are sent to. Since the row.ColumnsChannelMiddleware
 	// instance that sends columns to this channel is running before the row firmware, we should be careful
@@ -196,7 +196,8 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 
 	if err != nil {
 		log.Debug().Err(err).Msg("error executing middlewares")
-		g := &errgroup.Group{}
+		g := &safegroup.Group{}
+
 		g.Go(func() error {
 			if dt_.JSStream != nil {
 				log.Debug().Msg("Closing JS stream")
@@ -206,21 +207,19 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 				log.Debug().Msg("Closing HTML stream")
 				close(dt_.HTMLStream)
 			}
-			if dt_.ErrorStream != nil {
-				v_ := err.Error()
-				log.Debug().Str("errorString", v_).Msg("Sending error to error stream")
-				select {
-				case dt_.ErrorStream <- v_:
-					log.Debug().Msg("Error sent to error stream")
-				default:
-					log.Debug().Msg("Error stream full")
-				}
-				log.Debug().Msg("Closing error stream")
-				close(dt_.ErrorStream)
+			v_ := err.Error()
+			log.Debug().Str("errorString", v_).Msg("Sending error to error stream")
+			select {
+			case dt_.ErrorStream <- v_:
+				log.Debug().Msg("Error sent to error stream")
+			default:
+				log.Debug().Msg("Error stream full")
 			}
+			log.Debug().Msg("Closing error stream")
 
 			return nil
 		})
+
 		g.Go(func() error {
 			log.Debug().Msg("Closing columns channel")
 			columnsC <- []types.FieldName{}
@@ -230,6 +229,7 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 			log.Debug().Msg("Closing columns channel")
 			return nil
 		})
+
 		g.Go(func() error {
 			log.Debug().Msg("Rendering template")
 			return qh.renderTemplate(parsedLayers, c.Response(), dt_, columnsC)
@@ -271,8 +271,7 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 	}
 
 	ctx := c.Request().Context()
-	ctx2, cancel := context.WithCancel(ctx)
-	eg, ctx3 := errgroup.WithContext(ctx2)
+	eg, ctx3 := safegroup.WithContext(ctx)
 
 	// copy the json rows to the template stream
 	eg.Go(func() error {
@@ -304,32 +303,27 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 
 	// actually run the command
 	eg.Go(func() error {
-		defer func() {
-			if dt_.ErrorStream != nil {
-				close(dt_.ErrorStream)
-				dt_.ErrorStream = nil
-			}
-			_ = cancel
-		}()
-
 		// NOTE(manuel, 2023-10-16) The GetAllParameterValues is a bit of a hack because really what we want is to only get those flags through the layers
 		err = qh.cmd.RunIntoGlazeProcessor(ctx3, parsedLayers, gp)
 
-		g := &errgroup.Group{}
-		if err != nil {
-			g.Go(func() error {
-				// make sure to render the ErrorStream at the bottom, because we would otherwise get into a deadlock with the streaming channels
-				// NOTE(manuel, 2024-05-14) I'm not sure if with the addition of goroutines this is actually still necessary
-				//
-				dt_.ErrorStream <- err.Error()
-				close(dt_.ErrorStream)
-				dt_.ErrorStream = nil
-				return nil
-			})
-		}
+		g := &safegroup.Group{}
+		err_ := err
+		g.Go(func() error {
+			log.Debug().Err(err_).Msg("error running command")
+			// make sure to render the ErrorStream at the bottom, because we would otherwise get into a deadlock with the streaming channels
+			// NOTE(manuel, 2024-05-14) I'm not sure if with the addition of goroutines this is actually still necessary
+			//
+			log.Debug().Msg("sending error to error stream")
+			if err_ != nil {
+				dt_.ErrorStream <- err_.Error()
+			}
+			close(dt_.ErrorStream)
+			return nil
+		})
 
 		g.Go(func() error {
-			err = gp.Close(ctx3)
+			err := gp.Close(ctx3)
+			log.Debug().Msg("closed gp")
 			if err != nil {
 				return err
 			}
@@ -340,20 +334,31 @@ func (qh *QueryHandler) Handle(c echo.Context) error {
 			return nil
 		})
 
-		return g.Wait()
+		err := g.Wait()
+		if err != nil {
+			return err
+		}
+		return nil
 	})
 
 	eg.Go(func() error {
 		// if qh.Cmd implements cmds.CommandWithMetadata, get Metadata
+		log.Debug().Msg("rendering template")
 		err := qh.renderTemplate(parsedLayers, c.Response(), dt_, columnsC)
+		log.Debug().Msg("rendered template")
 		if err != nil {
 			return err
 		}
 
 		return nil
 	})
 
-	return eg.Wait()
+	err = eg.Wait()
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (qh *QueryHandler) renderTemplate(

pkg/glazed/handlers/sse/sse.go

@@ -12,8 +12,8 @@ import (
 	"github.com/go-go-golems/glazed/pkg/middlewares/row"
 	"github.com/go-go-golems/parka/pkg/glazed/handlers"
 	middlewares2 "github.com/go-go-golems/parka/pkg/glazed/middlewares"
+	"github.com/kucherenkovova/safegroup"
 	"github.com/labstack/echo/v4"
-	"golang.org/x/sync/errgroup"
 	"net/http"
 )
 
@@ -65,7 +65,7 @@ func (h *QueryHandler) Handle(c echo.Context) error {
 		// to the client
 		sseWriter := NewSSEWriter()
 
-		eg := errgroup.Group{}
+		eg := safegroup.Group{}
 		eg.Go(func() error {
 			defer sseWriter.Close()
 			return cmd.RunIntoWriter(
@@ -107,7 +107,7 @@ func (h *QueryHandler) Handle(c echo.Context) error {
 		r := row.NewOutputChannelMiddleware(json2.NewOutputFormatter(), eventChan)
 		gp.AddRowMiddleware(r)
 
-		eg := errgroup.Group{}
+		eg := safegroup.Group{}
 		eg.Go(func() error {
 			err := cmd.RunIntoGlazeProcessor(ctx, parsedLayers, gp)
 			if err != nil {

pkg/server/server.go

@@ -178,6 +178,7 @@ func NewServer(options ...ServerOption) (*Server, error) {
 
 	router.Logger = lecho.From(log.Logger)
 
+	router.Use(middleware.Recover())
 	// Custom middleware logger using zerolog
 	router.Use(middleware.RequestLoggerWithConfig(middleware.RequestLoggerConfig{
 		LogURI:    true,