Merge pull request #6531 from tonistiigi/exec-order-fix

gateway: fix exec process lifecycle ordering

Author: tonistiigi

frontend/gateway/gateway.go

@@ -1571,12 +1571,20 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
 					return stack.Enable(err)
 				})
 
+				// startedSent gates the proc.Wait goroutine until
+				// Started is sent and output readers are spawned,
+				// preventing Exit-before-Started and a deadlock
+				// where pio.Close() races with reader setup.
+				startedSent := make(chan struct{})
+
 				eg.Go(func() error {
 					defer func() {
 						pio.Close()
 					}()
 					err := proc.Wait()
 
+					<-startedSent
+
 					var statusCode uint32
 					var exitError *pb.ExitError
 					var statusError *spb.Status
@@ -1626,6 +1634,7 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
 					},
 				})
 				if err != nil {
+					close(startedSent)
 					return stack.Enable(err)
 				}
 
@@ -1669,6 +1678,7 @@ func (lbf *llbBridgeForwarder) ExecProcess(srv pb.LLBBridge_ExecProcessServer) e
 						return stack.Enable(err)
 					})
 				}
+				close(startedSent)
 			}
 		}
 	})

frontend/gateway/grpcclient/client.go

@@ -820,9 +820,8 @@ func (c *grpcClient) Inputs(ctx context.Context) (map[string]llb.State, error) {
 // communication channel between the process and the ExecProcess message
 // stream.
 type procMessageForwarder struct {
-	done      chan struct{}
-	closeOnce sync.Once
-	msgs      chan *pb.ExecMessage
+	done chan struct{}
+	msgs chan *pb.ExecMessage
 }
 
 func newProcMessageForwarder() *procMessageForwarder {
@@ -836,9 +835,6 @@ func (b *procMessageForwarder) Send(ctx context.Context, m *pb.ExecMessage) {
 	select {
 	case <-ctx.Done():
 	case <-b.done:
-		b.closeOnce.Do(func() {
-			close(b.msgs)
-		})
 	case b.msgs <- m:
 	}
 }
@@ -856,8 +852,6 @@ func (b *procMessageForwarder) Recv(ctx context.Context) (m *pb.ExecMessage, ok
 
 func (b *procMessageForwarder) Close() {
 	close(b.done)
-	b.Recv(context.Background())      // flush any messages in queue
-	b.Send(context.Background(), nil) // ensure channel is closed
 }
 
 // messageForwarder manages a single grpc stream for ExecProcess to facilitate