solve: lock local cache ingests across clients

Independent local content stores only coordinate writers within each
store instance. Concurrent cache exports to the same directory can open
the same ingest files, causing rename failures or incomplete writes.

Acquire a filesystem lock per ingest reference before opening a writer
or aborting an ingest. Release it when the writer closes or commits,
including error paths, and report contention as unavailable so existing
retry logic can wait. Preserve references so interrupted uploads remain
resumable, and retain lock files to avoid races with replacement inodes.

Signed-off-by: CrazyMax <1951866+crazy-max@users.noreply.github.com>

client/client_cache_test.go

@@ -21,9 +21,11 @@ import (
 	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	ctd "github.com/containerd/containerd/v2/client"
+	"github.com/containerd/containerd/v2/core/content"
 	"github.com/containerd/containerd/v2/core/images"
 	ctdcompression "github.com/containerd/containerd/v2/pkg/archive/compression"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
+	contentlocal "github.com/containerd/containerd/v2/plugins/content/local"
 	cerrdefs "github.com/containerd/errdefs"
 	"github.com/gofrs/flock"
 	cacheimporttypes "github.com/moby/buildkit/cache/remotecache/v1/types"
@@ -31,6 +33,8 @@ import (
 	"github.com/moby/buildkit/exporter/containerimage/exptypes"
 	gateway "github.com/moby/buildkit/frontend/gateway/client"
 	"github.com/moby/buildkit/identity"
+	"github.com/moby/buildkit/session"
+	sessioncontent "github.com/moby/buildkit/session/content"
 	"github.com/moby/buildkit/util/compression"
 	"github.com/moby/buildkit/util/testutil"
 	"github.com/moby/buildkit/util/testutil/helpers"
@@ -40,8 +44,147 @@ import (
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
 	"github.com/tonistiigi/fsutil"
+	"golang.org/x/sync/errgroup"
 )
 
+func testConcurrentLocalCacheExport(t *testing.T, sb integration.Sandbox) {
+	integration.SkipOnPlatform(t, "windows")
+	workers.CheckFeatureCompat(t, sb, workers.FeatureCacheExport, workers.FeatureCacheBackendLocal)
+
+	ctx, cancel := context.WithTimeoutCause(sb.Context(), time.Minute, errors.New("concurrent local cache export timed out"))
+	defer cancel()
+
+	c, err := New(ctx, sb.Address())
+	require.NoError(t, err)
+	defer c.Close()
+
+	// Export a single scratch-mounted layer in the default min mode so both
+	// exports contend on the same ref before either writes config or manifest.
+	st := llb.Image("busybox:latest").Run(llb.Shlex("sh -c 'echo shared > /out/payload'")).AddMount("/out", llb.Scratch())
+	def, err := st.Marshal(ctx)
+	require.NoError(t, err)
+
+	dir := t.TempDir()
+	opened := make(chan struct{}, 2)
+	blocked := make(chan struct{}, 1)
+	releases := [2]chan struct{}{make(chan struct{}), make(chan struct{})}
+	completed := make(chan error, 2)
+	eg, ctx := errgroup.WithContext(ctx)
+	for i := range 2 {
+		store, err := newLocalCacheStore(dir)
+		require.NoError(t, err)
+		s, err := session.NewSession(ctx, "")
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, s.Close()) })
+		s.Allow(sessioncontent.NewAttachable(map[string]content.Store{
+			"local:" + dir: &gatedCacheStore{Store: store, opened: opened, blocked: blocked, release: releases[i]},
+		}))
+		eg.Go(func() error { return s.Run(ctx, c.Dialer()) })
+		eg.Go(func() error {
+			defer s.Close()
+			_, err := c.Solve(ctx, def, SolveOpt{
+				SharedSession:         s,
+				SessionPreInitialized: true,
+				CacheExports: []CacheOptionsEntry{{Type: "local", Attrs: map[string]string{
+					"dest": dir,
+					"tag":  fmt.Sprintf("export-%d", i),
+				}}},
+			}, nil)
+			completed <- err
+			return err
+		})
+		if i == 0 {
+			select {
+			case <-opened:
+			case <-ctx.Done():
+				t.Fatal(context.Cause(ctx))
+			}
+		}
+	}
+	done := make(chan error, 1)
+	go func() { done <- eg.Wait() }()
+
+	// The first export owns the ingest. The second must retry rather than
+	// opening and modifying the same temporary files.
+	select {
+	case <-blocked:
+	case <-opened:
+		t.Fatal("second export opened an active ingest")
+	case <-ctx.Done():
+		t.Fatal(context.Cause(ctx))
+	}
+
+	entries, err := os.ReadDir(filepath.Join(dir, "ingest"))
+	require.NoError(t, err)
+	require.Len(t, entries, 1, "exports of the same reference must reuse its ingest")
+	// Index updates use a non-blocking lock. Delay the second export's retry
+	// until the first has published its index entry.
+	for _, release := range releases {
+		close(release)
+		require.NoError(t, <-completed)
+	}
+	require.NoError(t, <-done)
+
+	dt, err := os.ReadFile(filepath.Join(dir, "index.json"))
+	require.NoError(t, err)
+	var index ocispecs.Index
+	require.NoError(t, json.Unmarshal(dt, &index))
+	require.Len(t, index.Manifests, 2)
+
+	store, err := contentlocal.NewStore(dir)
+	require.NoError(t, err)
+
+	for _, desc := range index.Manifests {
+		dt, err := content.ReadBlob(sb.Context(), store, desc)
+		require.NoError(t, err)
+		var manifest ocispecs.Manifest
+		require.NoError(t, json.Unmarshal(dt, &manifest))
+		for _, blob := range append(manifest.Layers, manifest.Config) {
+			dt, err := content.ReadBlob(sb.Context(), store, blob)
+			require.NoError(t, err)
+			require.Equal(t, blob.Digest, blob.Digest.Algorithm().FromBytes(dt))
+		}
+	}
+}
+
+// gatedCacheStore pauses the first writer so two exports can overlap without
+// relying on scheduling or a large payload to keep an ingest active.
+type gatedCacheStore struct {
+	content.Store
+	opened  chan<- struct{}
+	blocked chan<- struct{}
+	release <-chan struct{}
+	paused  atomic.Bool
+}
+
+func (s *gatedCacheStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {
+	w, err := s.Store.Writer(ctx, opts...)
+	if err != nil {
+		if errors.Is(err, cerrdefs.ErrUnavailable) {
+			select {
+			case s.blocked <- struct{}{}:
+			default:
+			}
+			select {
+			case <-s.release:
+			case <-ctx.Done():
+				return nil, context.Cause(ctx)
+			}
+		}
+		return nil, err
+	}
+	if s.paused.CompareAndSwap(false, true) {
+		s.opened <- struct{}{}
+		select {
+		case <-s.release:
+		case <-ctx.Done():
+			w.Close()
+			return nil, context.Cause(ctx)
+		}
+	}
+	return w, nil
+}
+
 func testBasicAzblobCacheImportExport(t *testing.T, sb integration.Sandbox) {
 	integration.SkipOnPlatform(t, "windows")
 	workers.CheckFeatureCompat(t, sb,

client/client_test.go

@@ -22,6 +22,7 @@ var allTests = []func(t *testing.T, sb integration.Sandbox){
 	testBasicInlineCacheImportExport,
 	testBasicLocalCacheImportExport,
 	testAlternativeInputKeysCacheImport,
+	testConcurrentLocalCacheExport,
 	testBasicRegistryCacheImportExport,
 	testBasicS3CacheImportExport,
 	testCacheExportCacheDeletedContent,

client/solve.go

@@ -17,6 +17,7 @@ import (
 	"github.com/containerd/containerd/v2/core/content"
 	"github.com/containerd/containerd/v2/core/images"
 	contentlocal "github.com/containerd/containerd/v2/plugins/content/local"
+	cerrdefs "github.com/containerd/errdefs"
 	"github.com/gofrs/flock"
 	controlapi "github.com/moby/buildkit/api/services/control"
 	"github.com/moby/buildkit/client/llb"
@@ -640,7 +641,7 @@ func parseCacheOptions(ctx context.Context, isGateway bool, opt SolveOpt) (*cach
 			if err := os.MkdirAll(csDir, 0755); err != nil {
 				return nil, err
 			}
-			cs, err := contentlocal.NewStore(csDir)
+			cs, err := newLocalCacheStore(csDir)
 			if err != nil {
 				return nil, err
 			}
@@ -686,7 +687,7 @@ func parseCacheOptions(ctx context.Context, isGateway bool, opt SolveOpt) (*cach
 			if csDir == "" {
 				return nil, errors.New("local cache importer requires src")
 			}
-			cs, err := contentlocal.NewStore(csDir)
+			cs, err := newLocalCacheStore(csDir)
 			if err != nil {
 				bklog.G(ctx).Warning("local cache import at " + csDir + " not found due to err: " + err.Error())
 				continue
@@ -752,3 +753,92 @@ func parseCacheOptions(ctx context.Context, isGateway bool, opt SolveOpt) (*cach
 	}
 	return &res, nil
 }
+
+// localCacheStore coordinates ingests across clients sharing a cache directory.
+// The underlying local store only locks references within one store instance.
+type localCacheStore struct {
+	content.Store
+	dir string
+}
+
+func newLocalCacheStore(dir string) (content.Store, error) {
+	store, err := contentlocal.NewStore(dir)
+	if err != nil {
+		return nil, err
+	}
+	return &localCacheStore{Store: store, dir: dir}, nil
+}
+
+func (s *localCacheStore) lock(ref string) (*flock.Flock, error) {
+	dir := filepath.Join(s.dir, "ingest-locks")
+	if err := os.MkdirAll(dir, 0755); err != nil {
+		return nil, err
+	}
+	// Keep lock files after unlocking: deleting one could let another client
+	// lock a different inode while an existing waiter still uses the old one.
+	// This retains one empty file per ref, even after cache reset. The reset
+	// lock cannot protect their removal: not all users of this store hold it.
+	lock := flock.New(filepath.Join(dir, digest.FromString(ref).Encoded()))
+	locked, err := lock.TryLock()
+	if err != nil {
+		return nil, err
+	}
+	if !locked {
+		return nil, errors.Wrapf(cerrdefs.ErrUnavailable, "ingest ref %q is locked", ref)
+	}
+	return lock, nil
+}
+
+func (s *localCacheStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {
+	var wo content.WriterOpts
+	for _, opt := range opts {
+		if err := opt(&wo); err != nil {
+			return nil, err
+		}
+	}
+	if wo.Ref == "" {
+		return nil, errors.Wrap(cerrdefs.ErrInvalidArgument, "ref must not be empty")
+	}
+	lock, err := s.lock(wo.Ref)
+	if err != nil {
+		return nil, err
+	}
+	w, err := s.Store.Writer(ctx, content.WithRef(wo.Ref), content.WithDescriptor(wo.Desc))
+	if err != nil {
+		lock.Close()
+		return nil, err
+	}
+	return &localCacheWriter{Writer: w, lock: lock}, nil
+}
+
+func (s *localCacheStore) Abort(ctx context.Context, ref string) error {
+	lock, err := s.lock(ref)
+	if err != nil {
+		return err
+	}
+	defer lock.Close()
+	return s.Store.Abort(ctx, ref)
+}
+
+type localCacheWriter struct {
+	content.Writer
+	lock *flock.Flock
+}
+
+func (w *localCacheWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {
+	err := w.Writer.Commit(ctx, size, expected, opts...)
+	closeErr := w.Close()
+	if err != nil {
+		return err
+	}
+	return closeErr
+}
+
+func (w *localCacheWriter) Close() error {
+	err := w.Writer.Close()
+	lockErr := w.lock.Close()
+	if err != nil {
+		return err
+	}
+	return lockErr
+}

client/solve_test.go

@@ -1,12 +1,100 @@
 package client
 
 import (
+	"context"
+	"os"
+	"path/filepath"
 	"testing"
 
+	"github.com/containerd/containerd/v2/core/content"
+	cerrdefs "github.com/containerd/errdefs"
 	"github.com/moby/buildkit/client/llb"
+	digest "github.com/opencontainers/go-digest"
+	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
 )
 
+func TestLocalCacheStoreLockAndResume(t *testing.T) {
+	ctx := t.Context()
+	dir := t.TempDir()
+	first, err := newLocalCacheStore(dir)
+	require.NoError(t, err)
+	second, err := newLocalCacheStore(dir)
+	require.NoError(t, err)
+	data := []byte("resumable cache layer")
+	desc := ocispecs.Descriptor{Digest: digest.FromBytes(data), Size: int64(len(data))}
+	opts := []content.WriterOpt{content.WithRef("layer"), content.WithDescriptor(desc)}
+	w, err := first.Writer(ctx, opts...)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, w.Close()) })
+	_, err = w.Write(data[:5])
+	require.NoError(t, err)
+
+	_, err = second.Writer(ctx, opts...)
+	require.ErrorIs(t, err, cerrdefs.ErrUnavailable)
+	require.ErrorIs(t, second.Abort(ctx, "layer"), cerrdefs.ErrUnavailable)
+	// Cancellation of OpenWriter's retry must not disturb the active writer.
+	cancelled, cancel := context.WithCancelCause(ctx)
+	cancel(errors.New("cancel retry"))
+	_, err = content.OpenWriter(cancelled, second, opts...)
+	require.ErrorIs(t, err, cerrdefs.ErrUnavailable)
+
+	// Independent references can still be written concurrently.
+	other, err := second.Writer(ctx, content.WithRef("other"))
+	require.NoError(t, err)
+	require.NoError(t, other.Close())
+	require.NoError(t, second.Abort(ctx, "other"))
+	require.NoError(t, w.Close())
+
+	resumed, err := second.Writer(ctx, opts...)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, resumed.Close()) })
+	status, err := resumed.Status()
+	require.NoError(t, err)
+	require.EqualValues(t, 5, status.Offset)
+	_, err = resumed.Write(data[5:])
+	require.NoError(t, err)
+	require.NoError(t, resumed.Commit(ctx, desc.Size, desc.Digest))
+	blob, err := content.ReadBlob(ctx, first, desc)
+	require.NoError(t, err)
+	require.Equal(t, data, blob)
+	entries, err := os.ReadDir(filepath.Join(dir, "ingest"))
+	require.NoError(t, err)
+	require.Empty(t, entries, "resuming an interrupted export must reclaim its ingest")
+
+	// An already-existing blob must not leave its reference locked.
+	_, err = first.Writer(ctx, opts...)
+	require.ErrorIs(t, err, cerrdefs.ErrAlreadyExists)
+	require.NoError(t, second.Abort(ctx, "layer"))
+}
+
+func TestLocalCacheStoreCommitError(t *testing.T) {
+	for _, failure := range []string{"size", "option"} {
+		t.Run(failure, func(t *testing.T) {
+			ctx := t.Context()
+			dir := t.TempDir()
+			first, err := newLocalCacheStore(dir)
+			require.NoError(t, err)
+			second, err := newLocalCacheStore(dir)
+			require.NoError(t, err)
+			w, err := first.Writer(ctx, content.WithRef("layer"))
+			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, w.Close()) })
+			if failure == "size" {
+				err = w.Commit(ctx, 1, "")
+				require.ErrorIs(t, err, cerrdefs.ErrFailedPrecondition)
+			} else {
+				sentinel := errors.New("invalid commit option")
+				err = w.Commit(ctx, 0, "", func(*content.Info) error { return sentinel })
+				require.ErrorIs(t, err, sentinel)
+			}
+			// Commit closes the writer even on failure, including the file lock.
+			require.NoError(t, second.Abort(ctx, "layer"))
+		})
+	}
+}
+
 func TestSolveRejectsInvalidLocalExporterMode(t *testing.T) {
 	st := llb.Scratch().File(
 		llb.Mkfile("fresh.txt", 0600, []byte("fresh")),