dap: make exec shell persistent across the build

Invoking the shell will cause it to persist across the entire build and
to re-execute whenever the builder pauses at another location again.

This still requires using `exec` to launch the shell. Launching by frame
id is also removed since it no longer applies to this version.

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

build/result.go

@@ -1,10 +1,15 @@
 package build
 
 import (
+	"cmp"
 	"context"
 	_ "crypto/sha256" // ensure digests can be computed
 	"encoding/json"
 	"io"
+	iofs "io/fs"
+	"path/filepath"
+	"slices"
+	"strings"
 	"sync"
 
 	"github.com/moby/buildkit/exporter/containerimage/exptypes"
@@ -14,6 +19,7 @@ import (
 	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"github.com/tonistiigi/fsutil/types"
 )
 
 // NewResultHandle stores a gateway client, gateway reference, and the error from
@@ -75,6 +81,40 @@ func (r *ResultHandle) NewContainer(ctx context.Context, cfg *InvokeConfig) (gat
 	return r.gwClient.NewContainer(ctx, req)
 }
 
+func (r *ResultHandle) StatFile(ctx context.Context, fpath string, cfg *InvokeConfig) (*types.Stat, error) {
+	containerCfg, err := r.getContainerConfig(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	candidateMounts := make([]gateway.Mount, 0, len(containerCfg.Mounts))
+	for _, m := range containerCfg.Mounts {
+		if strings.HasPrefix(fpath, m.Dest) {
+			candidateMounts = append(candidateMounts, m)
+		}
+	}
+	if len(candidateMounts) == 0 {
+		return nil, iofs.ErrNotExist
+	}
+
+	slices.SortFunc(candidateMounts, func(a, b gateway.Mount) int {
+		return cmp.Compare(len(a.Dest), len(b.Dest))
+	})
+
+	m := candidateMounts[len(candidateMounts)-1]
+	relpath, err := filepath.Rel(m.Dest, fpath)
+	if err != nil {
+		return nil, err
+	}
+
+	if m.Ref == nil {
+		return nil, iofs.ErrNotExist
+	}
+
+	req := gateway.StatRequest{Path: filepath.ToSlash(relpath)}
+	return m.Ref.StatFile(ctx, req)
+}
+
 func (r *ResultHandle) getContainerConfig(cfg *InvokeConfig) (containerCfg gateway.NewContainerRequest, _ error) {
 	if r.ref != nil && r.solveErr == nil {
 		logrus.Debugf("creating container from successful build")

dap/adapter.go

@@ -38,9 +38,14 @@ type Adapter[C LaunchConfig] struct {
 	threadsMu    sync.RWMutex
 	nextThreadID int
 
+	sharedState
+}
+
+type sharedState struct {
 	breakpointMap *breakpointMap
-	sourceMap     sourceMap
+	sourceMap     *sourceMap
 	idPool        *idPool
+	sh            *shell
 }
 
 func New[C LaunchConfig]() *Adapter[C] {
@@ -51,8 +56,12 @@ func New[C LaunchConfig]() *Adapter[C] {
 		evaluateReqCh: make(chan *evaluateRequest),
 		threads:       make(map[int]*thread),
 		nextThreadID:  1,
-		breakpointMap: newBreakpointMap(),
+		sharedState: sharedState{
-		idPool:        new(idPool),
+			breakpointMap: newBreakpointMap(),
+			sourceMap:     new(sourceMap),
+			idPool:        new(idPool),
+			sh:            newShell(),
+		},
 	}
 	d.srv = NewServer(d.dapHandler())
 	return d
@@ -233,12 +242,10 @@ func (d *Adapter[C]) newThread(ctx Context, name string) (t *thread) {
 	d.threadsMu.Lock()
 	id := d.nextThreadID
 	t = &thread{
-		id:            id,
+		id:          id,
-		name:          name,
+		name:        name,
-		sourceMap:     &d.sourceMap,
+		sharedState: d.sharedState,
-		breakpointMap: d.breakpointMap,
+		variables:   newVariableReferences(),
-		idPool:        d.idPool,
-		variables:     newVariableReferences(),
 	}
 	d.threads[t.id] = t
 	d.nextThreadID++
@@ -261,20 +268,6 @@ func (d *Adapter[C]) getThread(id int) (t *thread) {
 	return t
 }
 
-func (d *Adapter[C]) getFirstThread() (t *thread) {
-	d.threadsMu.Lock()
-	defer d.threadsMu.Unlock()
-
-	for _, thread := range d.threads {
-		if thread.isPaused() {
-			if t == nil || thread.id < t.id {
-				t = thread
-			}
-		}
-	}
-	return t
-}
-
 func (d *Adapter[C]) deleteThread(ctx Context, t *thread) {
 	d.threadsMu.Lock()
 	if t := d.threads[t.id]; t != nil {

dap/debug_shell.go

@@ -0,0 +1,308 @@
+package dap
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"io/fs"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+
+	"github.com/docker/buildx/build"
+	"github.com/docker/buildx/util/ioset"
+	"github.com/docker/cli/cli-plugins/metadata"
+	"github.com/google/go-dap"
+	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
+)
+
+type shell struct {
+	// SocketPath is set on the first time Init is invoked
+	// and stays that way.
+	SocketPath string
+
+	// Locks access to the session from the debug adapter.
+	// Only one debug thread can access the shell at a time.
+	sem *semaphore.Weighted
+
+	// Initialized once per shell and reused.
+	once sync.Once
+	err  error
+	l    net.Listener
+	eg   *errgroup.Group
+
+	// For the specific session.
+	fwd       *ioset.Forwarder
+	connected chan struct{}
+	mu        sync.RWMutex
+}
+
+func newShell() *shell {
+	sh := &shell{
+		sem: semaphore.NewWeighted(1),
+	}
+	sh.resetSession()
+	return sh
+}
+
+func (s *shell) resetSession() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.fwd = nil
+	s.connected = make(chan struct{})
+}
+
+// Init initializes the shell for connections on the client side.
+// Attach will block until the terminal has been initialized.
+func (s *shell) Init() error {
+	return s.listen()
+}
+
+func (s *shell) listen() error {
+	s.once.Do(func() {
+		var dir string
+		dir, s.err = os.MkdirTemp("", "buildx-dap-exec")
+		if s.err != nil {
+			return
+		}
+		defer func() {
+			if s.err != nil {
+				os.RemoveAll(dir)
+			}
+		}()
+		s.SocketPath = filepath.Join(dir, "s.sock")
+
+		s.l, s.err = net.Listen("unix", s.SocketPath)
+		if s.err != nil {
+			return
+		}
+
+		s.eg, _ = errgroup.WithContext(context.Background())
+		s.eg.Go(s.acceptLoop)
+	})
+	return s.err
+}
+
+func (s *shell) acceptLoop() error {
+	for {
+		if err := s.accept(); err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				return nil
+			}
+			return err
+		}
+	}
+}
+
+func (s *shell) accept() error {
+	conn, err := s.l.Accept()
+	if err != nil {
+		return err
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.fwd != nil {
+		writeLine(conn, "Error: Already connected to exec instance.")
+		conn.Close()
+		return nil
+	}
+
+	// Set the input of the forwarder to the connection.
+	s.fwd = ioset.NewForwarder()
+	s.fwd.SetIn(&ioset.In{
+		Stdin:  io.NopCloser(conn),
+		Stdout: conn,
+		Stderr: nopCloser{conn},
+	})
+
+	close(s.connected)
+	writeLine(conn, "Attached to build process.")
+	return nil
+}
+
+// Attach will attach the given thread to the shell.
+// Only one container can attach to a shell at any given time.
+// Other attaches will block until the context is canceled or it is
+// able to reserve the shell for its own use.
+//
+// This method is intended to be called by paused threads.
+func (s *shell) Attach(ctx context.Context, t *thread) {
+	rCtx := t.rCtx
+	if rCtx == nil {
+		return
+	}
+
+	var f dap.StackFrame
+	if len(t.stackTrace) > 0 {
+		f = t.frames[t.stackTrace[0]].StackFrame
+	}
+
+	cfg := &build.InvokeConfig{Tty: true}
+	if len(cfg.Entrypoint) == 0 && len(cfg.Cmd) == 0 {
+		cfg.Entrypoint = []string{"/bin/sh"} // launch shell by default
+		cfg.Cmd = []string{}
+		cfg.NoCmd = false
+	}
+
+	for {
+		if err := s.attach(ctx, f, rCtx, cfg); err != nil {
+			return
+		}
+	}
+}
+
+func (s *shell) wait(ctx context.Context) error {
+	s.mu.RLock()
+	connected := s.connected
+	s.mu.RUnlock()
+
+	select {
+	case <-connected:
+		return nil
+	case <-ctx.Done():
+		return context.Cause(ctx)
+	}
+}
+
+func (s *shell) attach(ctx context.Context, f dap.StackFrame, rCtx *build.ResultHandle, cfg *build.InvokeConfig) (retErr error) {
+	if err := s.wait(ctx); err != nil {
+		return err
+	}
+
+	in, out := ioset.Pipe()
+	defer in.Close()
+	defer out.Close()
+
+	s.mu.RLock()
+	fwd := s.fwd
+	s.mu.RUnlock()
+
+	fwd.SetOut(&out)
+	defer func() {
+		if retErr != nil {
+			fwd.SetOut(nil)
+		}
+	}()
+
+	// Check if the entrypoint is executable. If it isn't, don't bother
+	// trying to invoke.
+	if reason, ok := s.canInvoke(ctx, rCtx, cfg); !ok {
+		writeLineF(in.Stdout, "Build container is not executable. (reason: %s)", reason)
+		<-ctx.Done()
+		return context.Cause(ctx)
+	}
+
+	if err := s.sem.Acquire(ctx, 1); err != nil {
+		return err
+	}
+	defer s.sem.Release(1)
+
+	ctr, err := build.NewContainer(ctx, rCtx, cfg)
+	if err != nil {
+		return err
+	}
+	defer ctr.Cancel()
+
+	writeLineF(in.Stdout, "Running %s in build container from line %d.",
+		strings.Join(append(cfg.Entrypoint, cfg.Cmd...), " "),
+		f.Line,
+	)
+
+	writeLine(in.Stdout, "Changes to the container will be reset after the next step is executed.")
+	err = ctr.Exec(ctx, cfg, in.Stdin, in.Stdout, in.Stderr)
+
+	// Send newline to properly terminate the output.
+	writeLine(in.Stdout, "")
+
+	if err != nil {
+		return err
+	}
+
+	fwd.Close()
+	s.resetSession()
+	return nil
+}
+
+func (s *shell) canInvoke(ctx context.Context, rCtx *build.ResultHandle, cfg *build.InvokeConfig) (reason string, ok bool) {
+	var cmd string
+	if len(cfg.Entrypoint) > 0 {
+		cmd = cfg.Entrypoint[0]
+	} else if len(cfg.Cmd) > 0 {
+		cmd = cfg.Cmd[0]
+	}
+
+	if cmd == "" {
+		return "no command specified", false
+	}
+
+	st, err := rCtx.StatFile(ctx, cmd, cfg)
+	if err != nil {
+		return fmt.Sprintf("stat error: %s", err), false
+	}
+
+	mode := fs.FileMode(st.Mode)
+	if !mode.IsRegular() {
+		return fmt.Sprintf("%s: not a file", cmd), false
+	}
+	if mode&0111 == 0 {
+		return fmt.Sprintf("%s: not an executable", cmd), false
+	}
+	return "", true
+}
+
+// SendRunInTerminalRequest will send the request to the client to attach to
+// the socket path that was created by Init. This is intended to be run
+// from the adapter and interact directly with the client.
+func (s *shell) SendRunInTerminalRequest(ctx Context) error {
+	// TODO: this should work in standalone mode too.
+	docker := os.Getenv(metadata.ReexecEnvvar)
+	req := &dap.RunInTerminalRequest{
+		Request: dap.Request{
+			Command: "runInTerminal",
+		},
+		Arguments: dap.RunInTerminalRequestArguments{
+			Kind: "integrated",
+			Args: []string{docker, "buildx", "dap", "attach", s.SocketPath},
+			Env: map[string]any{
+				"BUILDX_EXPERIMENTAL": "1",
+			},
+		},
+	}
+
+	resp := ctx.Request(req)
+	if !resp.GetResponse().Success {
+		return errors.New(resp.GetResponse().Message)
+	}
+	return nil
+}
+
+type nopCloser struct {
+	io.Writer
+}
+
+func (nopCloser) Close() error {
+	return nil
+}
+
+func writeLine(w io.Writer, msg string) {
+	if os.PathSeparator == '\\' {
+		fmt.Fprint(w, msg+"\r\n")
+	} else {
+		fmt.Fprintln(w, msg)
+	}
+}
+
+func writeLineF(w io.Writer, format string, a ...any) {
+	if os.PathSeparator == '\\' {
+		fmt.Fprintf(w, format+"\r\n", a...)
+	} else {
+		fmt.Fprintf(w, format+"\n", a...)
+	}
+}

dap/eval.go

@@ -30,7 +30,7 @@ func (d *Adapter[C]) Evaluate(ctx Context, req *dap.EvaluateRequest, resp *dap.E
 	}
 
 	var retErr error
-	cmd := d.replCommands(ctx, req, resp, &retErr)
+	cmd := d.replCommands(ctx, resp, &retErr)
 	cmd.SetArgs(args)
 	cmd.SetErr(d.Out())
 	if err := cmd.Execute(); err != nil {
@@ -42,39 +42,34 @@ func (d *Adapter[C]) Evaluate(ctx Context, req *dap.EvaluateRequest, resp *dap.E
 	return retErr
 }
 
-func (d *Adapter[C]) replCommands(ctx Context, req *dap.EvaluateRequest, resp *dap.EvaluateResponse, retErr *error) *cobra.Command {
+func (d *Adapter[C]) replCommands(ctx Context, resp *dap.EvaluateResponse, retErr *error) *cobra.Command {
 	rootCmd := &cobra.Command{
 		SilenceErrors: true,
 	}
 
-	execCmd, execOpts := replCmd(ctx, "exec", resp, retErr, d.execCmd)
+	execCmd, _ := replCmd(ctx, "exec", resp, retErr, d.execCmd)
-	execCmd.PreRun = func(cmd *cobra.Command, args []string) {
-		execOpts.FrameID = req.Arguments.FrameId
-	}
 	rootCmd.AddCommand(execCmd)
 	return rootCmd
 }
 
-type execOptions struct {
+type execOptions struct{}
-	FrameID int
-}
 
-func (d *Adapter[C]) execCmd(ctx Context, args []string, flags execOptions) (string, error) {
+func (d *Adapter[C]) execCmd(ctx Context, _ []string, _ execOptions) (string, error) {
 	if !d.supportsExec {
 		return "", errors.New("cannot exec without runInTerminal client capability")
 	}
 
-	var t *thread
+	// Initialize the shell if it hasn't been done before. This will allow any
-	if flags.FrameID > 0 {
+	// containers that are attempting to attach to actually attach.
-		if t = d.getThreadByFrameID(flags.FrameID); t == nil {
+	if err := d.sh.Init(); err != nil {
-			return "", errors.Errorf("no thread with frame id %d", flags.FrameID)
+		return "", err
-		}
-	} else {
-		if t = d.getFirstThread(); t == nil {
-			return "", errors.New("no paused thread")
-		}
 	}
-	return t.Exec(ctx, args)
+
+	// Send the request to attach to the terminal.
+	if err := d.sh.SendRunInTerminalRequest(ctx); err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("Started process attached to %s.", d.sh.SocketPath), nil
 }
 
 func replCmd[Flags any, RetVal any](ctx Context, name string, resp *dap.EvaluateResponse, retErr *error, fn func(ctx Context, args []string, flags Flags) (RetVal, error)) (*cobra.Command, *Flags) {

dap/thread.go

@@ -22,10 +22,8 @@ type thread struct {
 	name string
 
 	// Persistent state from the adapter.
-	idPool        *idPool
+	sharedState
-	sourceMap     *sourceMap
+	variables *variableReferences
-	breakpointMap *breakpointMap
-	variables     *variableReferences
 
 	// Inputs to the evaluate call.
 	c          gateway.Client
@@ -50,7 +48,7 @@ type thread struct {
 	mu     sync.Mutex
 
 	// Attributes set when a thread is paused.
-	cancel     context.CancelCauseFunc
+	cancel     context.CancelCauseFunc // invoked when the thread is resumed
 	rCtx       *build.ResultHandle
 	curPos     digest.Digest
 	stackTrace []int32
@@ -254,9 +252,6 @@ func (t *thread) pause(c Context, ref gateway.Reference, err error, pos *step, e
 	}
 
 	t.paused = make(chan stepType, 1)
-	if ref != nil || err != nil {
-		t.rCtx = build.NewResultHandle(c, t.c, ref, t.meta, err)
-	}
 	if err != nil {
 		var solveErr *errdefs.SolveError
 		if errors.As(err, &solveErr) {
@@ -270,6 +265,10 @@ func (t *thread) pause(c Context, ref gateway.Reference, err error, pos *step, e
 	t.collectStackTrace(ctx, pos, ref)
 	t.cancel = cancel
 
+	if ref != nil || err != nil {
+		t.prepareResultHandle(c, ref, err)
+	}
+
 	event.ThreadId = t.id
 	c.C() <- &dap.StoppedEvent{
 		Event: dap.Event{Event: "stopped"},
@@ -278,6 +277,27 @@ func (t *thread) pause(c Context, ref gateway.Reference, err error, pos *step, e
 	return t.paused
 }
 
+func (t *thread) prepareResultHandle(c Context, ref gateway.Reference, err error) {
+	// Create a context for cancellations and make the cancel function
+	// block on the wait group.
+	var wg sync.WaitGroup
+	ctx, cancel := context.WithCancelCause(c)
+	t.cancel = func(cause error) {
+		defer wg.Wait()
+		cancel(cause)
+	}
+
+	t.rCtx = build.NewResultHandle(ctx, t.c, ref, t.meta, err)
+
+	// Start the attach. Use the context we created and perform it in
+	// a goroutine. We aren't necessarily assuming this will actually work.
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		t.sh.Attach(ctx, t)
+	}()
+}
+
 func (t *thread) Continue() {
 	t.resume(stepContinue)
 }
@@ -308,13 +328,6 @@ func (t *thread) resume(step stepType) {
 	t.paused = nil
 }
 
-func (t *thread) isPaused() bool {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-
-	return t.paused != nil
-}
-
 func (t *thread) StackTrace() []dap.StackFrame {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -494,17 +507,20 @@ func (t *thread) solve(ctx context.Context, target digest.Digest) (gateway.Refer
 }
 
 func (t *thread) releaseState() {
-	if t.cancel != nil {
-		t.cancel(context.Canceled)
-		t.cancel = nil
-	}
 	if t.rCtx != nil {
 		t.rCtx.Done()
 		t.rCtx = nil
 	}
+
 	for _, f := range t.frames {
 		f.ResetVars()
 	}
+
+	if t.cancel != nil {
+		t.cancel(context.Canceled)
+		t.cancel = nil
+	}
+
 	t.stackTrace = t.stackTrace[:0]
 	t.variables.Reset()
 }