host: fix flynn run --enable-log hang caused by OOM logger WaitGroup leak

The OOM notification logger shared the container's MuxConfig (same JobID),
causing Follow() to add a 4th entry to the job's WaitGroup. Since cleanup()
only closes the 3 tracked stdio streams, the WaitGroup never reached zero,
blocking jobDoneCh() forever and preventing AttachExit from being sent.

Fix by giving the OOM logger a separate config with a distinct JobID so it
gets its own independent WaitGroup. Also fix the Attach deferred func to
handle the case where the container exits before client is set, and add a
jobDones map to logmux so late subscribers can detect already-finished jobs.

Fixes: TestRun, TestRunSignal, TestLog, TestLogFilter, TestLogFollow,
TestLogStderr (all were hanging indefinitely).

Author: root

controller/client/v1/client.go

@@ -624,7 +624,7 @@ func (c *Client) StreamJobEvents(appID string, output chan *ct.Job) (stream.Stre
 }
 
 func (c *Client) WatchJobEvents(appID, releaseID string) (ct.JobWatcher, error) {
-	events := make(chan *ct.Job)
+	events := make(chan *ct.Job, 100)
 	stream, err := c.StreamJobEvents(appID, events)
 	if err != nil {
 		return nil, err

host/libcontainer_backend.go

@@ -1161,7 +1161,11 @@ func (c *Container) watch(ready chan<- error, buffer host.LogBuffer) error {
 		log.Warn("error subscribing to OOM notifications (non-fatal)", "err", err)
 	} else {
 		go func() {
-			logger := c.l.LogMux.Logger(logagg.MsgIDInit, c.MuxConfig, "component", "flynn-host")
+			// Use a separate config so the OOM logger's Follow() does not
+			// add to the job's WaitGroup (which tracks stdio stream closure).
+			oomConfig := *c.MuxConfig
+			oomConfig.JobID = c.job.ID + ".oom"
+			logger := c.l.LogMux.Logger(logagg.MsgIDInit, &oomConfig, "component", "flynn-host")
 			defer logger.Close()
 			for range notifyOOM {
 				logger.Crit("FATAL: a container process was killed due to lack of available memory")
@@ -1397,14 +1401,21 @@ func (l *LibcontainerBackend) Attach(req *AttachRequest) (err error) {
 	}
 
 	defer func() {
-		if client != nil && (req.Job.Job.Config.TTY || req.Stream) && err == io.EOF {
-			<-client.done
+		if (req.Job.Job.Config.TTY || req.Stream) && err == io.EOF {
+			if client != nil {
+				<-client.done
+			}
 			job := l.State.GetJob(req.Job.Job.ID)
+			if job == nil {
+				return
+			}
 			if job.Status == host.StatusDone || job.Status == host.StatusCrashed {
 				err = ExitError(*job.ExitStatus)
 				return
 			}
-			err = errors.New(*job.Error)
+			if job.Error != nil {
+				err = errors.New(*job.Error)
+			}
 		}
 	}()
 
@@ -1485,8 +1496,10 @@ func (l *LibcontainerBackend) Attach(req *AttachRequest) (err error) {
 	}
 
 	ch := make(chan *rfc5424.Message)
+	l.Logger.Info("streaming log for attached job", "fn", "Attach", "job.id", req.Job.Job.ID, "logs", req.Logs, "stream", req.Stream)
 	stream, err := l.LogMux.StreamLog(req.Job.Job.Metadata["flynn-controller.app"], req.Job.Job.ID, req.Logs, req.Stream, ch)
 	if err != nil {
+		l.Logger.Error("StreamLog error", "fn", "Attach", "job.id", req.Job.Job.ID, "err", err)
 		return err
 	}
 	defer stream.Close()
@@ -1508,6 +1521,7 @@ func (l *LibcontainerBackend) Attach(req *AttachRequest) (err error) {
 			return nil
 		}
 	}
+	l.Logger.Info("log stream ended for attached job", "fn", "Attach", "job.id", req.Job.Job.ID)
 
 	return io.EOF
 }

host/logmux/logmux.go

@@ -44,6 +44,9 @@ type Mux struct {
 	// following a job, in order to wait on the appropriate group from jobWaits,
 	// as a WaitGroup can't be waited until the counter is >0
 	jobStarts map[string]chan struct{}
+	// jobDones tracks jobs whose log streams have all closed, so that
+	// late subscribers can detect that the job is already done
+	jobDones map[string]struct{}
 
 	subscribersMtx sync.RWMutex
 	subscribers    map[string]map[chan message]struct{}
@@ -61,6 +64,7 @@ func New(hostID, logDir string, logger log15.Logger) *Mux {
 		logger:      logger,
 		jobWaits:    make(map[string]*sync.WaitGroup),
 		jobStarts:   make(map[string]chan struct{}),
+		jobDones:    make(map[string]struct{}),
 		subscribers: make(map[string]map[chan message]struct{}),
 		appLogs:     make(map[string]*appLog),
 	}
@@ -319,6 +323,7 @@ func (m *Mux) Follow(r io.ReadCloser, buffer string, msgID logagg.MsgID, config
 			m.jobsMtx.Lock()
 			defer m.jobsMtx.Unlock()
 			delete(m.jobWaits, config.JobID)
+			m.jobDones[config.JobID] = struct{}{}
 		}()
 	}
 
@@ -403,14 +408,18 @@ func (m *Mux) StreamLog(appID, jobID string, history, follow bool, ch chan<- *rf
 }
 
 // jobDoneCh returns a channel that is closed when all of the streams we are
-// following from the job have been closed. It will never unblock if the job has
-// already finished.
+// following from the job have been closed.
 func (m *Mux) jobDoneCh(jobID string, stop <-chan struct{}) <-chan struct{} {
 	ch := make(chan struct{})
 	go func() {
 		defer close(ch)
 		var started chan struct{}
 		m.jobsMtx.Lock()
+		// check if the job has already completed
+		if _, done := m.jobDones[jobID]; done {
+			m.jobsMtx.Unlock()
+			return
+		}
 		// check if there is already a WaitGroup
 		wg, ok := m.jobWaits[jobID]
 		if !ok {
@@ -449,6 +458,7 @@ func (m *Mux) jobDoneCh(jobID string, stop <-chan struct{}) <-chan struct{} {
 }
 
 func (m *Mux) followLog(appID, jobID string, ch chan<- *rfc5424.Message) (stream.Stream, error) {
+	m.logger.Info("followLog starting", "app.id", appID, "job.id", jobID)
 	s := stream.New()
 	var jobDone <-chan struct{}
 	if jobID != "" {