build: change build handler to evaluate instead of onresult

This changes the build handler to customize the behavior of evaluate
rather than onresult and also simplifies the `ResultHandle`. The
`ResultHandle` is now only valid within the gateway callback and can be
used to start containers from the handler.

`Evaluate` now executes inside of the gateway callback rather than
having a separate implementation that executes or re-invokes the build.
This keeps the gateway callback session open until the debugger has
returned.

The `ErrReload` for monitor has now been moved into the `build` package
and been renamed to `ErrRestart`. This is because it restarts the build
so the name makes a bit more sense. The actual use of this functionality
is still tied to the monitor reload.

Signed-off-by: Jonathan A. Sternberg <jonathan.sternberg@docker.com>

build/build.go

@@ -59,6 +59,8 @@ const (
 	printLintFallbackImage = "docker/dockerfile:1.8.1@sha256:e87caa74dcb7d46cd820352bfea12591f3dba3ddc4285e19c7dcd13359f7cefd"
 )
 
+var ErrRestart = errors.New("build: restart")
+
 type Options struct {
 	Inputs Inputs
 
@@ -312,7 +314,7 @@ func toRepoOnly(in string) (string, error) {
 }
 
 type Handler struct {
-	OnResult func(driverIdx int, rCtx *ResultHandle)
+	Evaluate func(ctx context.Context, c gateway.Client, res *gateway.Result) error
 }
 
 func Build(ctx context.Context, nodes []builder.Node, opts map[string]Options, docker *dockerutil.Client, cfg *confutil.Config, w progress.Writer) (resp map[string]*client.SolveResponse, err error) {
@@ -479,9 +481,14 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[
 					ch, done := progress.NewChannel(pw)
 					defer func() { <-done }()
 
-					cc := c
+					var (
-					var callRes map[string][]byte
+						callRes     map[string][]byte
-					buildFunc := func(ctx context.Context, c gateway.Client) (*gateway.Result, error) {
+						frontendErr error
+					)
+					buildFunc := func(ctx context.Context, c gateway.Client) (_ *gateway.Result, retErr error) {
+						// Capture the error from this build function.
+						defer catchFrontendError(&retErr, &frontendErr)
+
 						if opt.CallFunc != nil {
 							if _, ok := req.FrontendOpt["frontend.caps"]; !ok {
 								req.FrontendOpt["frontend.caps"] = "moby.buildkit.frontend.subrequests+forward"
@@ -504,24 +511,25 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[
 						results.Set(rKey, res)
 
 						if children := childTargets[rKey]; len(children) > 0 {
-							if err := waitForChildren(ctx, res, results, children); err != nil {
+							if err := waitForChildren(ctx, bh, c, res, results, children); err != nil {
+								return nil, err
+							}
+						} else if bh != nil && bh.Evaluate != nil {
+							if err := bh.Evaluate(ctx, c, res); err != nil {
 								return nil, err
 							}
 						}
 						return res, nil
 					}
+
 					buildRef := fmt.Sprintf("%s/%s/%s", node.Builder, node.Name, so.Ref)
 
-					var rr *client.SolveResponse
+					span, ctx := tracing.StartSpan(ctx, "build")
-					if bh != nil && bh.OnResult != nil {
+					rr, err := c.Build(ctx, *so, "buildx", buildFunc, ch)
-						var resultHandle *ResultHandle
+					if errors.Is(frontendErr, ErrRestart) {
-						resultHandle, rr, err = NewResultHandle(ctx, cc, *so, "buildx", buildFunc, ch)
+						err = ErrRestart
-						bh.OnResult(dp.driverIndex, resultHandle)
-					} else {
-						span, ctx := tracing.StartSpan(ctx, "build")
-						rr, err = c.Build(ctx, *so, "buildx", buildFunc, ch)
-						tracing.FinishWithError(span, err)
 					}
+					tracing.FinishWithError(span, err)
 
 					if !so.Internal && desktop.BuildBackendEnabled() && node.Driver.HistoryAPISupported(ctx) {
 						if err != nil {
@@ -1191,7 +1199,7 @@ func solve(ctx context.Context, c gateway.Client, req gateway.SolveRequest) (*ga
 	return res, nil
 }
 
-func waitForChildren(ctx context.Context, res *gateway.Result, results *waitmap.Map, children []string) error {
+func waitForChildren(ctx context.Context, bh *Handler, c gateway.Client, res *gateway.Result, results *waitmap.Map, children []string) error {
 	// wait for the child targets to register their LLB before evaluating
 	_, err := results.Get(ctx, children...)
 	if err != nil {
@@ -1200,6 +1208,9 @@ func waitForChildren(ctx context.Context, res *gateway.Result, results *waitmap.
 	// we need to wait until the child targets have completed before we can release
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
+		if bh != nil && bh.Evaluate != nil {
+			return bh.Evaluate(ctx, c, res)
+		}
 		return res.EachRef(func(ref gateway.Reference) error {
 			return ref.Evaluate(ctx)
 		})
@@ -1210,3 +1221,12 @@ func waitForChildren(ctx context.Context, res *gateway.Result, results *waitmap.
 	})
 	return eg.Wait()
 }
+
+func catchFrontendError(retErr, frontendErr *error) {
+	*frontendErr = *retErr
+	if errors.Is(*retErr, ErrRestart) {
+		// Overwrite the sentinel error with a more user friendly message.
+		// This gets stored only in the return error.
+		*retErr = errors.New("build restarted by client")
+	}
+}

build/invoke.go

@@ -56,26 +56,18 @@ type Container struct {
 func NewContainer(ctx context.Context, resultCtx *ResultHandle, cfg *InvokeConfig) (*Container, error) {
 	mainCtx := ctx
 
-	ctrCh := make(chan *Container)
+	ctrCh := make(chan *Container, 1)
-	errCh := make(chan error)
+	errCh := make(chan error, 1)
 	go func() {
-		err := resultCtx.build(func(ctx context.Context, c gateway.Client) (*gateway.Result, error) {
+		err := func() error {
-			ctx, cancel := context.WithCancelCause(ctx)
-			go func() {
-				<-mainCtx.Done()
-				cancel(errors.WithStack(context.Canceled))
-			}()
-
-			containerCfg, err := resultCtx.getContainerConfig(cfg)
-			if err != nil {
-				return nil, err
-			}
 			containerCtx, containerCancel := context.WithCancelCause(ctx)
 			defer containerCancel(errors.WithStack(context.Canceled))
-			bkContainer, err := c.NewContainer(containerCtx, containerCfg)
+
+			bkContainer, err := resultCtx.NewContainer(containerCtx, cfg)
 			if err != nil {
-				return nil, err
+				return err
 			}
+
 			releaseCh := make(chan struct{})
 			container := &Container{
 				containerCancel: containerCancel,
@@ -92,8 +84,8 @@ func NewContainer(ctx context.Context, resultCtx *ResultHandle, cfg *InvokeConfi
 			ctrCh <- container
 			<-container.releaseCh
 
-			return nil, bkContainer.Release(ctx)
+			return bkContainer.Release(ctx)
-		})
+		}()
 		if err != nil {
 			errCh <- err
 		}

build/result.go

@@ -7,259 +7,41 @@ import (
 	"io"
 	"sync"
 
-	"github.com/moby/buildkit/client"
 	"github.com/moby/buildkit/exporter/containerimage/exptypes"
 	gateway "github.com/moby/buildkit/frontend/gateway/client"
 	"github.com/moby/buildkit/solver/errdefs"
 	"github.com/moby/buildkit/solver/pb"
-	"github.com/moby/buildkit/solver/result"
 	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/errgroup"
 )
 
-// NewResultHandle makes a call to client.Build, additionally returning a
+// NewResultHandle stores a gateway client, gateway result, and the error from
-// opaque ResultHandle alongside the standard response and error.
+// an evaluate call if it is present.
 //
 // This ResultHandle can be used to execute additional build steps in the same
 // context as the build occurred, which can allow easy debugging of build
 // failures and successes.
 //
 // If the returned ResultHandle is not nil, the caller must call Done() on it.
-func NewResultHandle(ctx context.Context, cc *client.Client, opt client.SolveOpt, product string, buildFunc gateway.BuildFunc, ch chan *client.SolveStatus) (*ResultHandle, *client.SolveResponse, error) {
+func NewResultHandle(ctx context.Context, c gateway.Client, res *gateway.Result, err error) *ResultHandle {
-	// Create a new context to wrap the original, and cancel it when the
+	rCtx := &ResultHandle{
-	// caller-provided context is cancelled.
+		res:      res,
-	//
+		gwClient: c,
-	// We derive the context from the background context so that we can forbid
-	// cancellation of the build request after <-done is closed (which we do
-	// before returning the ResultHandle).
-	baseCtx := ctx
-	ctx, cancel := context.WithCancelCause(context.Background())
-	done := make(chan struct{})
-	go func() {
-		select {
-		case <-baseCtx.Done():
-			cancel(baseCtx.Err())
-		case <-done:
-			// Once done is closed, we've recorded a ResultHandle, so we
-			// shouldn't allow cancelling the underlying build request anymore.
-		}
-	}()
-
-	// Create a new channel to forward status messages to the original.
-	//
-	// We do this so that we can discard status messages after the main portion
-	// of the build is complete. This is necessary for the solve error case,
-	// where the original gateway is kept open until the ResultHandle is
-	// closed - we don't want progress messages from operations in that
-	// ResultHandle to display after this function exits.
-	//
-	// Additionally, callers should wait for the progress channel to be closed.
-	// If we keep the session open and never close the progress channel, the
-	// caller will likely hang.
-	baseCh := ch
-	ch = make(chan *client.SolveStatus)
-	go func() {
-		for {
-			s, ok := <-ch
-			if !ok {
-				return
-			}
-			select {
-			case <-baseCh:
-				// base channel is closed, discard status messages
-			default:
-				baseCh <- s
-			}
-		}
-	}()
-	defer close(baseCh)
-
-	var resp *client.SolveResponse
-	var respErr error
-	var respHandle *ResultHandle
-
-	go func() {
-		defer func() { cancel(errors.WithStack(context.Canceled)) }() // ensure no dangling processes
-
-		var res *gateway.Result
-		var err error
-		resp, err = cc.Build(ctx, opt, product, func(ctx context.Context, c gateway.Client) (*gateway.Result, error) {
-			var err error
-			res, err = buildFunc(ctx, c)
-
-			if res != nil && err == nil {
-				// Force evaluation of the build result (otherwise, we likely
-				// won't get a solve error)
-				def, err2 := getDefinition(ctx, res)
-				if err2 != nil {
-					return nil, err2
-				}
-				res, err = evalDefinition(ctx, c, def)
-			}
-
-			if err != nil {
-				// Scenario 1: we failed to evaluate a node somewhere in the
-				// build graph.
-				//
-				// In this case, we construct a ResultHandle from this
-				// original Build session, and return it alongside the original
-				// build error. We then need to keep the gateway session open
-				// until the caller explicitly closes the ResultHandle.
-
-				var se *errdefs.SolveError
-				if errors.As(err, &se) {
-					respHandle = &ResultHandle{
-						done:     make(chan struct{}),
-						solveErr: se,
-						gwClient: c,
-						gwCtx:    ctx,
-					}
-					respErr = err // return original error to preserve stacktrace
-					close(done)
-
-					// Block until the caller closes the ResultHandle.
-					select {
-					case <-respHandle.done:
-					case <-ctx.Done():
-					}
-				}
-			}
-			return res, err
-		}, ch)
-		if respHandle != nil {
-			return
-		}
-		if err != nil {
-			// Something unexpected failed during the build, we didn't succeed,
-			// but we also didn't make it far enough to create a ResultHandle.
-			respErr = err
-			close(done)
-			return
-		}
-
-		// Scenario 2: we successfully built the image with no errors.
-		//
-		// In this case, the original gateway session has now been closed
-		// since the Build has been completed. So, we need to create a new
-		// gateway session to populate the ResultHandle. To do this, we
-		// need to re-evaluate the target result, in this new session. This
-		// should be instantaneous since the result should be cached.
-
-		def, err := getDefinition(ctx, res)
-		if err != nil {
-			respErr = err
-			close(done)
-			return
-		}
-
-		// NOTE: ideally this second connection should be lazily opened
-		opt := opt
-		opt.Ref = ""
-		opt.Exports = nil
-		opt.CacheExports = nil
-		opt.Internal = true
-		_, respErr = cc.Build(ctx, opt, "buildx", func(ctx context.Context, c gateway.Client) (*gateway.Result, error) {
-			res, err := evalDefinition(ctx, c, def)
-			if err != nil {
-				// This should probably not happen, since we've previously
-				// successfully evaluated the same result with no issues.
-				return nil, errors.Wrap(err, "inconsistent solve result")
-			}
-			respHandle = &ResultHandle{
-				done:     make(chan struct{}),
-				res:      res,
-				gwClient: c,
-				gwCtx:    ctx,
-			}
-			close(done)
-
-			// Block until the caller closes the ResultHandle.
-			select {
-			case <-respHandle.done:
-			case <-ctx.Done():
-			}
-			return nil, context.Cause(ctx)
-		}, nil)
-		if respHandle != nil {
-			return
-		}
-		close(done)
-	}()
-
-	// Block until the other thread signals that it's completed the build.
-	select {
-	case <-done:
-	case <-baseCtx.Done():
-		if respErr == nil {
-			respErr = baseCtx.Err()
-		}
 	}
-	return respHandle, resp, respErr
+	if err != nil && !errors.As(err, &rCtx.solveErr) {
-}
-
-// getDefinition converts a gateway result into a collection of definitions for
-// each ref in the result.
-func getDefinition(ctx context.Context, res *gateway.Result) (*result.Result[*pb.Definition], error) {
-	return result.ConvertResult(res, func(ref gateway.Reference) (*pb.Definition, error) {
-		st, err := ref.ToState()
-		if err != nil {
-			return nil, err
-		}
-		def, err := st.Marshal(ctx)
-		if err != nil {
-			return nil, err
-		}
-		return def.ToPB(), nil
-	})
-}
-
-// evalDefinition performs the reverse of getDefinition, converting a
-// collection of definitions into a gateway result.
-func evalDefinition(ctx context.Context, c gateway.Client, defs *result.Result[*pb.Definition]) (*gateway.Result, error) {
-	// force evaluation of all targets in parallel
-	results := make(map[*pb.Definition]*gateway.Result)
-	resultsMu := sync.Mutex{}
-	eg, egCtx := errgroup.WithContext(ctx)
-	defs.EachRef(func(def *pb.Definition) error {
-		eg.Go(func() error {
-			res, err := c.Solve(egCtx, gateway.SolveRequest{
-				Evaluate:   true,
-				Definition: def,
-			})
-			if err != nil {
-				return err
-			}
-			resultsMu.Lock()
-			results[def] = res
-			resultsMu.Unlock()
-			return nil
-		})
 		return nil
-	})
-	if err := eg.Wait(); err != nil {
-		return nil, err
 	}
-	res, _ := result.ConvertResult(defs, func(def *pb.Definition) (gateway.Reference, error) {
+	return rCtx
-		if res, ok := results[def]; ok {
-			return res.Ref, nil
-		}
-		return nil, nil
-	})
-	return res, nil
 }
 
 // ResultHandle is a build result with the client that built it.
 type ResultHandle struct {
 	res      *gateway.Result
 	solveErr *errdefs.SolveError
-
-	done     chan struct{}
-	doneOnce sync.Once
-
 	gwClient gateway.Client
-	gwCtx    context.Context
+
+	doneOnce sync.Once
 
 	cleanups   []func()
 	cleanupsMu sync.Mutex
@@ -274,9 +56,6 @@ func (r *ResultHandle) Done() {
 		for _, f := range cleanups {
 			f()
 		}
-
-		close(r.done)
-		<-r.gwCtx.Done()
 	})
 }
 
@@ -286,9 +65,12 @@ func (r *ResultHandle) registerCleanup(f func()) {
 	r.cleanupsMu.Unlock()
 }
 
-func (r *ResultHandle) build(buildFunc gateway.BuildFunc) (err error) {
+func (r *ResultHandle) NewContainer(ctx context.Context, cfg *InvokeConfig) (gateway.Container, error) {
-	_, err = buildFunc(r.gwCtx, r.gwClient)
+	req, err := r.getContainerConfig(cfg)
-	return err
+	if err != nil {
+		return nil, err
+	}
+	return r.gwClient.NewContainer(ctx, req)
 }
 
 func (r *ResultHandle) getContainerConfig(cfg *InvokeConfig) (containerCfg gateway.NewContainerRequest, _ error) {

commands/build.go

@@ -430,23 +430,11 @@ func runBuildWithOptions(ctx context.Context, dockerCli command.Cli, opts *Build
 	for {
 		resp, inputs, err := RunBuild(ctx, dockerCli, opts, in, printer, &bh)
 		if err != nil {
-			var be *BuildError
+			if errors.Is(err, build.ErrRestart) {
-			if errors.As(err, &be) {
+				retErr = nil
-				retErr = err
+				continue
-				// We can proceed to monitor
-			} else {
-				return nil, nil, errors.Wrapf(err, "failed to build")
-			}
-		}
-
-		if m != nil {
-			if err := m.Run(ctx, err); err != nil {
-				if errors.Is(err, monitor.ErrReload) {
-					retErr = nil
-					continue
-				}
-				logrus.Warnf("failed to run monitor: %v", err)
 			}
+			return nil, nil, errors.Wrapf(err, "failed to build")
 		}
 
 		return resp, inputs, err
@@ -1229,29 +1217,10 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *BuildOptions, inSt
 	resp, err := build.BuildWithResultHandler(ctx, nodes, buildOptions, dockerutil.NewClient(dockerCli), confutil.NewConfig(dockerCli), progress, bh)
 	err = wrapBuildError(err, false)
 	if err != nil {
-		return nil, nil, WrapBuild(err)
+		return nil, nil, err
 	}
 	if i, ok := buildOptions[defaultTargetName]; ok {
 		inputs = &i.Inputs
 	}
 	return resp[defaultTargetName], inputs, nil
 }
-
-type BuildError struct {
-	err error
-}
-
-func (e *BuildError) Unwrap() error {
-	return e.err
-}
-
-func (e *BuildError) Error() string {
-	return e.err.Error()
-}
-
-func WrapBuild(err error) error {
-	if err == nil {
-		return nil
-	}
-	return &BuildError{err: err}
-}

monitor/monitor.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"text/tabwriter"
+	"time"
 
 	"github.com/containerd/console"
 	"github.com/docker/buildx/build"
@@ -18,6 +19,7 @@ import (
 	"github.com/docker/buildx/util/ioset"
 	"github.com/docker/buildx/util/progress"
 	"github.com/google/shlex"
+	gateway "github.com/moby/buildkit/frontend/gateway/client"
 	"github.com/moby/buildkit/identity"
 	"github.com/moby/buildkit/solver/errdefs"
 	"github.com/pkg/errors"
@@ -34,10 +36,6 @@ type Monitor struct {
 	stdin  *ioset.SingleForwarder
 	stdout io.WriteCloser
 	stderr io.WriteCloser
-
-	res *build.ResultHandle
-	idx int
-	mu  sync.Mutex
 }
 
 func New(cfg *build.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCloser, printer *progress.Printer) *Monitor {
@@ -54,27 +52,38 @@ func New(cfg *build.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCl
 
 func (m *Monitor) Handler() build.Handler {
 	return build.Handler{
-		OnResult: func(driverIndex int, gotRes *build.ResultHandle) {
+		Evaluate: m.Evaluate,
-			m.mu.Lock()
-			defer m.mu.Unlock()
-
-			if m.res == nil || driverIndex < m.idx {
-				m.idx, m.res = driverIndex, gotRes
-			}
-		},
 	}
 }
 
-func (m *Monitor) Run(ctx context.Context, buildErr error) error {
+func (m *Monitor) Evaluate(ctx context.Context, c gateway.Client, res *gateway.Result) error {
-	defer m.reset()
+	buildErr := res.EachRef(func(ref gateway.Reference) error {
+		return ref.Evaluate(ctx)
+	})
 
-	if !m.invokeConfig.NeedsDebug(buildErr) {
+	if m.invokeConfig.NeedsDebug(buildErr) {
-		return nil
+		// Allow some time to ensure status updates are sent.
+		time.Sleep(200 * time.Millisecond)
+
+		// Print errors before launching monitor
+		if err := printError(buildErr, m.printer); err != nil {
+			logrus.Warnf("failed to print error information: %v", err)
+		}
+
+		rCtx := build.NewResultHandle(ctx, c, res, buildErr)
+		if monitorErr := m.Run(ctx, rCtx); monitorErr != nil {
+			if errors.Is(monitorErr, build.ErrRestart) {
+				return build.ErrRestart
+			}
+			logrus.Warnf("failed to run monitor: %v", monitorErr)
+		}
 	}
+	return buildErr
+}
 
-	// Print errors before launching monitor
+func (m *Monitor) Run(ctx context.Context, rCtx *build.ResultHandle) error {
-	if err := printError(buildErr, m.printer); err != nil {
+	if rCtx != nil {
-		logrus.Warnf("failed to print error information: %v", err)
+		defer rCtx.Done()
 	}
 
 	pr, pw := io.Pipe()
@@ -89,24 +98,13 @@ func (m *Monitor) Run(ctx context.Context, buildErr error) error {
 	}
 	defer con.Reset()
 
-	monitorErr := RunMonitor(ctx, m.invokeConfig, m.res, pr, m.stdout, m.stderr, m.printer)
+	monitorErr := RunMonitor(ctx, m.invokeConfig, rCtx, pr, m.stdout, m.stderr, m.printer)
 	if err := pw.Close(); err != nil {
 		logrus.Debug("failed to close monitor stdin pipe reader")
 	}
 	return monitorErr
 }
 
-func (m *Monitor) reset() {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	m.idx = 0
-	if m.res != nil {
-		m.res.Done()
-		m.res = nil
-	}
-}
-
 func (m *Monitor) Close() error {
 	return m.stdin.Close()
 }
@@ -380,7 +378,7 @@ func (m *monitor) Detach() {
 }
 
 func (m *monitor) Reload() {
-	m.cancel(ErrReload)
+	m.cancel(build.ErrRestart)
 }
 
 func (m *monitor) AttachedPID() string {