diff --git a/build/build.go b/build/build.go index 6c20d0eb3..12006a9da 100644 --- a/build/build.go +++ b/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 callRes map[string][]byte - buildFunc := func(ctx context.Context, c gateway.Client) (*gateway.Result, error) { + var ( + callRes map[string][]byte + 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 - if bh != nil && bh.OnResult != nil { - var resultHandle *ResultHandle - resultHandle, rr, err = NewResultHandle(ctx, cc, *so, "buildx", buildFunc, ch) - 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) + span, ctx := tracing.StartSpan(ctx, "build") + rr, err := c.Build(ctx, *so, "buildx", buildFunc, ch) + if errors.Is(frontendErr, ErrRestart) { + err = ErrRestart } + 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") + } +} diff --git a/build/invoke.go b/build/invoke.go index 5075530b3..1b79a987c 100644 --- a/build/invoke.go +++ b/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) - errCh := make(chan error) + ctrCh := make(chan *Container, 1) + errCh := make(chan error, 1) go func() { - err := resultCtx.build(func(ctx context.Context, c gateway.Client) (*gateway.Result, 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 - } + err := func() error { 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 } diff --git a/build/result.go b/build/result.go index d18f4a0b3..8e999944a 100644 --- a/build/result.go +++ b/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 -// opaque ResultHandle alongside the standard response and error. +// NewResultHandle stores a gateway client, gateway result, and the error from +// 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) { - // Create a new context to wrap the original, and cancel it when the - // caller-provided context is cancelled. - // - // 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() - } +func NewResultHandle(ctx context.Context, c gateway.Client, res *gateway.Result, err error) *ResultHandle { + rCtx := &ResultHandle{ + res: res, + gwClient: c, } - return respHandle, resp, respErr -} - -// 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 - }) + if err != nil && !errors.As(err, &rCtx.solveErr) { return nil - }) - if err := eg.Wait(); err != nil { - return nil, err } - res, _ := result.ConvertResult(defs, func(def *pb.Definition) (gateway.Reference, error) { - if res, ok := results[def]; ok { - return res.Ref, nil - } - return nil, nil - }) - return res, nil + return rCtx } // 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) { - _, err = buildFunc(r.gwCtx, r.gwClient) - return err +func (r *ResultHandle) NewContainer(ctx context.Context, cfg *InvokeConfig) (gateway.Container, error) { + req, err := r.getContainerConfig(cfg) + if err != nil { + return nil, err + } + return r.gwClient.NewContainer(ctx, req) } func (r *ResultHandle) getContainerConfig(cfg *InvokeConfig) (containerCfg gateway.NewContainerRequest, _ error) { diff --git a/commands/build.go b/commands/build.go index a4c4e2588..e1b8f86b6 100644 --- a/commands/build.go +++ b/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.As(err, &be) { - retErr = err - // 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) + if errors.Is(err, build.ErrRestart) { + retErr = nil + continue } + 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} -} diff --git a/monitor/monitor.go b/monitor/monitor.go index 09901291c..6981c5ccb 100644 --- a/monitor/monitor.go +++ b/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) { - m.mu.Lock() - defer m.mu.Unlock() - - if m.res == nil || driverIndex < m.idx { - m.idx, m.res = driverIndex, gotRes - } - }, + Evaluate: m.Evaluate, } } -func (m *Monitor) Run(ctx context.Context, buildErr error) error { - defer m.reset() +func (m *Monitor) Evaluate(ctx context.Context, c gateway.Client, res *gateway.Result) error { + buildErr := res.EachRef(func(ref gateway.Reference) error { + return ref.Evaluate(ctx) + }) - if !m.invokeConfig.NeedsDebug(buildErr) { - return nil + if m.invokeConfig.NeedsDebug(buildErr) { + // 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 - if err := printError(buildErr, m.printer); err != nil { - logrus.Warnf("failed to print error information: %v", err) +func (m *Monitor) Run(ctx context.Context, rCtx *build.ResultHandle) error { + if rCtx != nil { + defer rCtx.Done() } pr, pw := io.Pipe() @@ -89,34 +98,21 @@ 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() } // RunMonitor provides an interactive session for running and managing containers via specified IO. func RunMonitor(ctx context.Context, invokeConfig *build.InvokeConfig, rCtx *build.ResultHandle, stdin io.ReadCloser, stdout, stderr io.WriteCloser, progress *progress.Printer) error { - if err := progress.Pause(); err != nil { - return err - } - defer progress.Unpause() + progress.Pause() + defer progress.Resume() defer stdin.Close() @@ -382,7 +378,7 @@ func (m *monitor) Detach() { } func (m *monitor) Reload() { - m.cancel(ErrReload) + m.cancel(build.ErrRestart) } func (m *monitor) AttachedPID() string { @@ -475,10 +471,10 @@ func printError(err error, printer *progress.Printer) error { if err == nil { return nil } - if err := printer.Pause(); err != nil { - return err - } - defer printer.Unpause() + + printer.Pause() + defer printer.Resume() + for _, s := range errdefs.Sources(err) { s.Print(os.Stderr) } diff --git a/util/progress/printer.go b/util/progress/printer.go index 482641a18..85208b7e2 100644 --- a/util/progress/printer.go +++ b/util/progress/printer.go @@ -16,12 +16,24 @@ import ( "go.opentelemetry.io/otel/metric" ) +type printerState int + +const ( + printerStateDone printerState = iota + printerStateRunning + printerStatePaused +) + type Printer struct { - status chan *client.SolveStatus + out console.File + mode progressui.DisplayMode + opt *printerOpts + + status chan *client.SolveStatus + interrupt chan interruptRequest + state printerState - ready chan struct{} done chan struct{} - paused chan struct{} closeOnce sync.Once err error @@ -54,13 +66,23 @@ func (p *Printer) IsDone() bool { } func (p *Printer) Pause() error { - p.paused = make(chan struct{}) - return p.Wait() + done := make(chan struct{}) + p.interrupt <- interruptRequest{ + desiredState: printerStatePaused, + done: done, + } + + // Need to wait for a response to confirm we have control + // of the console output. + <-done + return nil } -func (p *Printer) Unpause() { - close(p.paused) - <-p.ready +func (p *Printer) Resume() { + p.interrupt <- interruptRequest{ + desiredState: printerStateRunning, + } + // Do not care about waiting for a response. } func (p *Printer) Write(s *client.SolveStatus) { @@ -115,42 +137,114 @@ func NewPrinter(ctx context.Context, out console.File, mode progressui.DisplayMo } pw := &Printer{ - ready: make(chan struct{}), - metrics: opt.mw, + out: out, + mode: mode, + opt: opt, + status: make(chan *client.SolveStatus), + interrupt: make(chan interruptRequest), + state: printerStateRunning, + done: make(chan struct{}), } + go pw.run(ctx, d) + + return pw, nil +} + +func (p *Printer) run(ctx context.Context, d progressui.Display) { + defer close(p.done) + defer close(p.interrupt) + + var ss []*client.SolveStatus + for p.state != printerStateDone { + switch p.state { + case printerStatePaused: + ss, p.err = p.bufferDisplay(ctx, ss) + case printerStateRunning: + var warnings []client.VertexWarning + warnings, ss, p.err = p.updateDisplay(ctx, d, ss) + p.warnings = append(p.warnings, warnings...) + + d, _ = p.newDisplay() + } + } + + if p.opt.onclose != nil { + p.opt.onclose() + } +} + +func (p *Printer) newDisplay() (progressui.Display, error) { + return progressui.NewDisplay(p.out, p.mode, p.opt.displayOpts...) +} + +func (p *Printer) updateDisplay(ctx context.Context, d progressui.Display, ss []*client.SolveStatus) ([]client.VertexWarning, []*client.SolveStatus, error) { + p.logMu.Lock() + p.logSourceMap = map[digest.Digest]any{} + p.logMu.Unlock() + + resumeLogs := logutil.Pause(logrus.StandardLogger()) + defer resumeLogs() + + interruptCh := make(chan interruptRequest, 1) + ingress := make(chan *client.SolveStatus) + go func() { + defer close(ingress) + defer close(interruptCh) + + for _, s := range ss { + ingress <- s + } + for { - pw.status = make(chan *client.SolveStatus) - pw.done = make(chan struct{}) - pw.closeOnce = sync.Once{} - - pw.logMu.Lock() - pw.logSourceMap = map[digest.Digest]any{} - pw.logMu.Unlock() - - resumeLogs := logutil.Pause(logrus.StandardLogger()) - close(pw.ready) - // not using shared context to not disrupt display but let is finish reporting errors - pw.warnings, pw.err = d.UpdateFrom(ctx, pw.status) - resumeLogs() - close(pw.done) - - if opt.onclose != nil { - opt.onclose() + select { + case s, ok := <-p.status: + if !ok { + return + } + ingress <- s + case req := <-p.interrupt: + interruptCh <- req + return + case <-ctx.Done(): + return } - if pw.paused == nil { - break - } - - pw.ready = make(chan struct{}) - <-pw.paused - pw.paused = nil - - d, _ = progressui.NewDisplay(out, mode, opt.displayOpts...) } }() - <-pw.ready - return pw, nil + + warnings, err := d.UpdateFrom(context.Background(), ingress) + if err == nil { + err = context.Cause(ctx) + } + + interrupt := <-interruptCh + p.state = interrupt.desiredState + interrupt.close() + return warnings, nil, err +} + +// bufferDisplay will buffer display updates from the status channel into a +// slice. +// +// This method returns if either status gets closed or if an interrupt is received. +func (p *Printer) bufferDisplay(ctx context.Context, ss []*client.SolveStatus) ([]*client.SolveStatus, error) { + for { + select { + case s, ok := <-p.status: + if !ok { + p.state = printerStateDone + return ss, nil + } + ss = append(ss, s) + case req := <-p.interrupt: + p.state = req.desiredState + req.close() + return ss, nil + case <-ctx.Done(): + p.state = printerStateDone + return nil, context.Cause(ctx) + } + } } func (p *Printer) WriteBuildRef(target string, ref string) { @@ -221,3 +315,14 @@ func dedupWarnings(inp []client.VertexWarning) []client.VertexWarning { } return res } + +type interruptRequest struct { + desiredState printerState + done chan<- struct{} +} + +func (req *interruptRequest) close() { + if req.done != nil { + close(req.done) + } +}