another try to remove job observer leaks

handlers/job/log.go

@@ -10,9 +10,9 @@ import (
 
 func (s *Handler) log(c *gin.Context) {
 	ctx, cancel := context.WithCancel(c.Request.Context())
+	defer cancel()
 	l, err := s.q.GetOrCreate(c.Param("queue_id")).Log(ctx, c.Param("job_id"))
 	if err != nil {
-		cancel()
 		_ = c.AbortWithError(http.StatusInternalServerError, err)
 		return
 	}
@@ -21,19 +21,13 @@ func (s *Handler) log(c *gin.Context) {
 	c.Header("Cache-Control", "no-cache,no-store,no-transform")
 	c.Header("Connection", "keep-alive")
 	c.Header("Access-Control-Allow-Origin", "*")
-	clientGone := c.Writer.CloseNotify()
 
 	c.Stream(func(w io.Writer) bool {
 		select {
-		case <-clientGone:
-			cancel()
-			return false
 		case <-c.Request.Context().Done():
-			cancel()
 			return false
 		case msg, ok := <-l:
 			if !ok {
-				cancel()
 				return false
 			}
 			c.SSEvent("message", msg)

handlers/job/script/action.go

@@ -224,8 +224,11 @@ func (s *ActionScript) warmUp(j *job.Job, m string, u string, su string, size in
 				log.WithError(err).Error("failed to get stats")
 				return
 			}
-			for ev := range ch {
+			select {
+			case ev := <-ch:
 				j.StatusUpdate(fmt.Sprintf("%v peers", ev.Peers))
+			case <-ctx2.Done():
+				return
 			}
 		}()
 	}

services/job/job.go

@@ -131,17 +131,11 @@ func (s *Job) Run(ctx context.Context) error {
 	return nil
 }
 
-func (s *Job) ObserveLog(ctx context.Context) *Observer {
+func (s *Job) ObserveLog() *Observer {
 	s.observersMux.Lock()
 	defer s.observersMux.Unlock()
 	o := NewObserver()
 	s.observers[o.ID] = o
-	go func(o *Observer, s *Job) {
-		<-ctx.Done()
-		s.observersMux.Lock()
-		defer s.observersMux.Unlock()
-		delete(s.observers, o.ID)
-	}(o, s)
 	return o
 }
 
@@ -359,7 +353,7 @@ func (s *Jobs) Log(ctx context.Context, id string) (c chan LogItem, err error) {
 		if j.closed {
 			close(c)
 		} else {
-			o := j.ObserveLog(ctx)
+			o := j.ObserveLog()
 			for {
 				select {
 				case <-ctx.Done():