llbsolver: move history blobs to a separate namespace

Migrate history objects to separate namespace to holding reference to a blob does not interfer with the GC labels held for same blobs by the containerd image store. Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>
2026-08-09 01:01:08 +00:00 · 2023-04-28 17:49:36 -07:00
parent 81d19ad945
commit f044e0a946
7 changed files with 288 additions and 51 deletions
--- a/control/control.go
+++ b/control/control.go
@@ -10,7 +10,6 @@ import (

 	contentapi "github.com/containerd/containerd/api/services/content/v1"
 	"github.com/containerd/containerd/content"
-	"github.com/containerd/containerd/leases"
 	"github.com/containerd/containerd/services/content/contentserver"
 	"github.com/docker/distribution/reference"
 	"github.com/mitchellh/hashstructure/v2"
@@ -26,12 +25,14 @@ import (
 	"github.com/moby/buildkit/frontend/attestations"
 	"github.com/moby/buildkit/session"
 	"github.com/moby/buildkit/session/grpchijack"
+	containerdsnapshot "github.com/moby/buildkit/snapshot/containerd"
 	"github.com/moby/buildkit/solver"
 	"github.com/moby/buildkit/solver/llbsolver"
 	"github.com/moby/buildkit/solver/llbsolver/proc"
 	"github.com/moby/buildkit/solver/pb"
 	"github.com/moby/buildkit/util/bklog"
 	"github.com/moby/buildkit/util/imageutil"
+	"github.com/moby/buildkit/util/leaseutil"
 	"github.com/moby/buildkit/util/throttle"
 	"github.com/moby/buildkit/util/tracing/transform"
 	"github.com/moby/buildkit/version"
@@ -58,8 +59,8 @@ type Opt struct {
 	Entitlements              []string
 	TraceCollector            sdktrace.SpanExporter
 	HistoryDB                 *bbolt.DB
-	LeaseManager              leases.Manager
-	ContentStore              content.Store
+	LeaseManager              *leaseutil.Manager
+	ContentStore              *containerdsnapshot.Store
 	HistoryConfig             *config.HistoryConfig
 }

@@ -79,12 +80,15 @@ type Controller struct { // TODO: ControlService
 func NewController(opt Opt) (*Controller, error) {
 	gatewayForwarder := controlgateway.NewGatewayForwarder()

-	hq := llbsolver.NewHistoryQueue(llbsolver.HistoryQueueOpt{
+	hq, err := llbsolver.NewHistoryQueue(llbsolver.HistoryQueueOpt{
 		DB:           opt.HistoryDB,
 		LeaseManager: opt.LeaseManager,
 		ContentStore: opt.ContentStore,
 		CleanConfig:  opt.HistoryConfig,
 	})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create history queue")
+	}

 	s, err := llbsolver.New(llbsolver.Opt{
 		WorkerController: opt.WorkerController,
@@ -125,7 +129,7 @@ func (c *Controller) Register(server *grpc.Server) {
 	c.gatewayForwarder.Register(server)
 	tracev1.RegisterTraceServiceServer(server, c)

-	store := &roContentStore{c.opt.ContentStore}
+	store := &roContentStore{c.opt.ContentStore.WithFallbackNS(c.opt.ContentStore.Namespace() + "_history")}
 	contentapi.RegisterContentServer(server, contentserver.New(store))
 }

--- a/snapshot/containerd/content.go
+++ b/snapshot/containerd/content.go
@@ -5,12 +5,15 @@ import (

 	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/namespaces"
+	"github.com/containerd/nydus-snapshotter/pkg/errdefs"
 	digest "github.com/opencontainers/go-digest"
 	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 )

-func NewContentStore(store content.Store, ns string) content.Store {
+type Store = nsContent
+
+func NewContentStore(store content.Store, ns string) *Store {
 	return &nsContent{ns, store}
 }

@@ -19,6 +22,14 @@ type nsContent struct {
 	content.Store
 }

+func (c *nsContent) Namespace() string {
+	return c.ns
+}
+
+func (c *nsContent) WithNamespace(ns string) *Store {
+	return NewContentStore(c.Store, ns)
+}
+
 func (c *nsContent) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
 	ctx = namespaces.WithNamespace(ctx, c.ns)
 	return c.Store.Info(ctx, dgst)
@@ -62,6 +73,13 @@ func (c *nsContent) Writer(ctx context.Context, opts ...content.WriterOpt) (cont
 	return c.writer(ctx, 3, opts...)
 }

+func (c *nsContent) WithFallbackNS(ns string) content.Store {
+	return &nsFallbackStore{
+		main: c,
+		fb:   c.WithNamespace(ns),
+	}
+}
+
 func (c *nsContent) writer(ctx context.Context, retries int, opts ...content.WriterOpt) (content.Writer, error) {
 	ctx = namespaces.WithNamespace(ctx, c.ns)
 	w, err := c.Store.Writer(ctx, opts...)
@@ -80,3 +98,58 @@ func (w *nsWriter) Commit(ctx context.Context, size int64, expected digest.Diges
 	ctx = namespaces.WithNamespace(ctx, w.ns)
 	return w.Writer.Commit(ctx, size, expected, opts...)
 }
+
+type nsFallbackStore struct {
+	main *nsContent
+	fb   *nsContent
+}
+
+var _ content.Store = &nsFallbackStore{}
+
+func (c *nsFallbackStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
+	info, err := c.main.Info(ctx, dgst)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			return c.fb.Info(ctx, dgst)
+		}
+	}
+	return info, err
+}
+
+func (c *nsFallbackStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) {
+	return c.main.Update(ctx, info, fieldpaths...)
+}
+
+func (c *nsFallbackStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error {
+	return c.main.Walk(ctx, fn, filters...)
+}
+
+func (c *nsFallbackStore) Delete(ctx context.Context, dgst digest.Digest) error {
+	return c.main.Delete(ctx, dgst)
+}
+
+func (c *nsFallbackStore) Status(ctx context.Context, ref string) (content.Status, error) {
+	return c.main.Status(ctx, ref)
+}
+
+func (c *nsFallbackStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) {
+	return c.main.ListStatuses(ctx, filters...)
+}
+
+func (c *nsFallbackStore) Abort(ctx context.Context, ref string) error {
+	return c.main.Abort(ctx, ref)
+}
+
+func (c *nsFallbackStore) ReaderAt(ctx context.Context, desc ocispecs.Descriptor) (content.ReaderAt, error) {
+	ra, err := c.main.ReaderAt(ctx, desc)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			return c.fb.ReaderAt(ctx, desc)
+		}
+	}
+	return ra, err
+}
+
+func (c *nsFallbackStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {
+	return c.main.Writer(ctx, opts...)
+}
--- a/solver/llbsolver/history.go
+++ b/solver/llbsolver/history.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"os"
 	"sort"
+	"strconv"
 	"sync"
 	"time"

@@ -16,6 +17,8 @@ import (
 	controlapi "github.com/moby/buildkit/api/services/control"
 	"github.com/moby/buildkit/client"
 	"github.com/moby/buildkit/cmd/buildkitd/config"
+	"github.com/moby/buildkit/identity"
+	containerdsnapshot "github.com/moby/buildkit/snapshot/containerd"
 	"github.com/moby/buildkit/util/leaseutil"
 	digest "github.com/opencontainers/go-digest"
 	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
@@ -25,24 +28,27 @@ import (

 const (
 	recordsBucket = "_records"
+	versionBucket = "_version"
 )

 type HistoryQueueOpt struct {
 	DB           *bolt.DB
-	LeaseManager leases.Manager
-	ContentStore content.Store
+	LeaseManager *leaseutil.Manager
+	ContentStore *containerdsnapshot.Store
 	CleanConfig  *config.HistoryConfig
 }

 type HistoryQueue struct {
 	// mu protects active, refs and deleted maps
-	mu       sync.Mutex
-	initOnce sync.Once
-	HistoryQueueOpt
-	ps      *pubsub[*controlapi.BuildHistoryEvent]
-	active  map[string]*controlapi.BuildHistoryRecord
-	refs    map[string]int
-	deleted map[string]struct{}
+	mu            sync.Mutex
+	initOnce      sync.Once
+	opt           HistoryQueueOpt
+	ps            *pubsub[*controlapi.BuildHistoryEvent]
+	active        map[string]*controlapi.BuildHistoryRecord
+	refs          map[string]int
+	deleted       map[string]struct{}
+	hContentStore *containerdsnapshot.Store
+	hLeaseManager *leaseutil.Manager
 }

 type StatusImportResult struct {
@@ -52,7 +58,7 @@ type StatusImportResult struct {
 	NumTotalSteps     int
 }

-func NewHistoryQueue(opt HistoryQueueOpt) *HistoryQueue {
+func NewHistoryQueue(opt HistoryQueueOpt) (*HistoryQueue, error) {
 	if opt.CleanConfig == nil {
 		opt.CleanConfig = &config.HistoryConfig{
 			MaxAge:     config.Duration{Duration: 48 * time.Hour},
@@ -60,7 +66,7 @@ func NewHistoryQueue(opt HistoryQueueOpt) *HistoryQueue {
 		}
 	}
 	h := &HistoryQueue{
-		HistoryQueueOpt: opt,
+		opt: opt,
 		ps: &pubsub[*controlapi.BuildHistoryEvent]{
 			m: map[*channel[*controlapi.BuildHistoryEvent]]struct{}{},
 		},
@@ -69,6 +75,39 @@ func NewHistoryQueue(opt HistoryQueueOpt) *HistoryQueue {
 		deleted: map[string]struct{}{},
 	}

+	ns := h.opt.ContentStore.Namespace()
+	// double check invalid configuration
+	ns2 := h.opt.LeaseManager.Namespace()
+	if ns != ns2 {
+		return nil, errors.Errorf("invalid configuration: content store namespace %q does not match lease manager namespace %q", ns, ns2)
+	}
+	h.hContentStore = h.opt.ContentStore.WithNamespace(ns + "_history")
+	h.hLeaseManager = h.opt.LeaseManager.WithNamespace(ns + "_history")
+
+	// v2 migration: all records need to be on isolated containerd ns from rest of buildkit
+	needsMigration := false
+	if err := h.opt.DB.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte(versionBucket))
+		if b != nil {
+			v := b.Get([]byte("version"))
+			if v != nil {
+				vi, err := strconv.ParseInt(string(v), 10, 64)
+				if err == nil && vi > 1 {
+					return nil
+				}
+			}
+		}
+		needsMigration = true
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	if needsMigration {
+		if err := h.migrateV2(); err != nil {
+			return nil, err
+		}
+	}
+
 	go func() {
 		for {
 			h.gc()
@@ -76,13 +115,108 @@ func NewHistoryQueue(opt HistoryQueueOpt) *HistoryQueue {
 		}
 	}()

-	return h
+	return h, nil
+}
+
+func (h *HistoryQueue) migrateV2() error {
+	ctx := context.Background()
+
+	if err := h.opt.DB.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte(recordsBucket))
+		if b == nil {
+			return nil
+		}
+		ctx, release, err := leaseutil.WithLease(ctx, h.hLeaseManager, leases.WithID("history_migration_"+identity.NewID()), leaseutil.MakeTemporary)
+		if err != nil {
+			return err
+		}
+		defer release(ctx)
+		return b.ForEach(func(key, dt []byte) error {
+			recs, err := h.opt.LeaseManager.ListResources(ctx, leases.Lease{ID: h.leaseID(string(key))})
+			if err != nil {
+				if errdefs.IsNotFound(err) {
+					return nil
+				}
+				return err
+			}
+			recs2 := make([]leases.Resource, 0, len(recs))
+			for _, r := range recs {
+				if r.Type == "content" {
+					if ok, err := h.migrateBlobV2(ctx, r.ID); err != nil {
+						return err
+					} else if ok {
+						recs2 = append(recs2, r)
+					}
+				} else {
+					return errors.Errorf("unknown resource type %q", r.Type)
+				}
+			}
+
+			l, err := h.hLeaseManager.Create(ctx, leases.WithID(h.leaseID(string(key))))
+			if err != nil {
+				if !errors.Is(err, errdefs.ErrAlreadyExists) {
+					return err
+				}
+				l = leases.Lease{ID: string(key)}
+			}
+
+			for _, r := range recs2 {
+				if err := h.hLeaseManager.AddResource(ctx, l, r); err != nil {
+					return err
+				}
+			}
+
+			return h.opt.LeaseManager.Delete(ctx, leases.Lease{ID: h.leaseID(string(key))})
+		})
+	}); err != nil {
+		return err
+	}
+
+	if err := h.opt.DB.Update(func(tx *bolt.Tx) error {
+		b, err := tx.CreateBucketIfNotExists([]byte(versionBucket))
+		if err != nil {
+			return err
+		}
+		return b.Put([]byte("version"), []byte("2"))
+	}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (h *HistoryQueue) migrateBlobV2(ctx context.Context, id string) (bool, error) {
+	dgst, err := digest.Parse(id)
+	if err != nil {
+		return false, err
+	}
+	w, err := content.OpenWriter(ctx, h.hContentStore, content.WithDescriptor(ocispecs.Descriptor{
+		Digest: dgst,
+	}), content.WithRef("history-migrate-"+id))
+	if err != nil {
+		if errdefs.IsAlreadyExists(err) {
+			return true, nil
+		}
+		return false, err
+	}
+	defer w.Close()
+	ra, err := h.opt.ContentStore.ReaderAt(ctx, ocispecs.Descriptor{
+		Digest: dgst,
+	})
+	if err != nil {
+		return false, nil // allow skipping
+	}
+	defer ra.Close()
+	if err := content.Copy(ctx, w, &reader{ReaderAt: ra}, 0, dgst); err != nil {
+		return false, err
+	}
+	return true, nil
 }

 func (h *HistoryQueue) gc() error {
 	var records []*controlapi.BuildHistoryRecord

-	if err := h.DB.View(func(tx *bolt.Tx) error {
+	if err := h.opt.DB.View(func(tx *bolt.Tx) error {
 		b := tx.Bucket([]byte(recordsBucket))
 		if b == nil {
 			return nil
@@ -103,7 +237,7 @@ func (h *HistoryQueue) gc() error {
 	}

 	// in order for record to get deleted by gc it exceed both maxentries and maxage criteria
-	if len(records) < int(h.CleanConfig.MaxEntries) {
+	if len(records) < int(h.opt.CleanConfig.MaxEntries) {
 		return nil
 	}

@@ -116,8 +250,8 @@ func (h *HistoryQueue) gc() error {
 	defer h.mu.Unlock()

 	now := time.Now()
-	for _, r := range records[h.CleanConfig.MaxEntries:] {
-		if now.Add(0 - h.CleanConfig.MaxAge.Duration).After(*r.CompletedAt) {
+	for _, r := range records[h.opt.CleanConfig.MaxEntries:] {
+		if now.Add(-h.opt.CleanConfig.MaxAge.Duration).After(*r.CompletedAt) {
 			if err := h.delete(r.Ref, false); err != nil {
 				return err
 			}
@@ -137,7 +271,7 @@ func (h *HistoryQueue) delete(ref string, sync bool) error {
 		Type:   controlapi.BuildHistoryEventType_DELETED,
 		Record: &controlapi.BuildHistoryRecord{Ref: ref},
 	})
-	if err := h.DB.Update(func(tx *bolt.Tx) error {
+	if err := h.opt.DB.Update(func(tx *bolt.Tx) error {
 		b := tx.Bucket([]byte(recordsBucket))
 		if b == nil {
 			return os.ErrNotExist
@@ -147,7 +281,7 @@ func (h *HistoryQueue) delete(ref string, sync bool) error {
 		if sync {
 			opts = append(opts, leases.SynchronousDelete)
 		}
-		err2 := h.LeaseManager.Delete(context.TODO(), leases.Lease{ID: h.leaseID(ref)}, opts...)
+		err2 := h.hLeaseManager.Delete(context.TODO(), leases.Lease{ID: h.leaseID(ref)}, opts...)
 		if err1 != nil {
 			return err1
 		}
@@ -161,7 +295,7 @@ func (h *HistoryQueue) delete(ref string, sync bool) error {
 func (h *HistoryQueue) init() error {
 	var err error
 	h.initOnce.Do(func() {
-		err = h.DB.Update(func(tx *bolt.Tx) error {
+		err = h.opt.DB.Update(func(tx *bolt.Tx) error {
 			_, err := tx.CreateBucketIfNotExists([]byte(recordsBucket))
 			return err
 		})
@@ -177,7 +311,23 @@ func (h *HistoryQueue) addResource(ctx context.Context, l leases.Lease, desc *co
 	if desc == nil {
 		return nil
 	}
-	return h.LeaseManager.AddResource(ctx, l, leases.Resource{
+	if _, err := h.hContentStore.Info(ctx, desc.Digest); err != nil {
+		if errdefs.IsNotFound(err) {
+			ctx, release, err := leaseutil.WithLease(ctx, h.hLeaseManager, leases.WithID("history_migration_"+identity.NewID()), leaseutil.MakeTemporary)
+			if err != nil {
+				return err
+			}
+			defer release(ctx)
+			ok, err := h.migrateBlobV2(ctx, string(desc.Digest))
+			if err != nil {
+				return err
+			}
+			if !ok {
+				return errors.Errorf("unknown blob %s in history", desc.Digest)
+			}
+		}
+	}
+	return h.hLeaseManager.AddResource(ctx, l, leases.Resource{
 		ID:   string(desc.Digest),
 		Type: "content",
 	})
@@ -188,7 +338,7 @@ func (h *HistoryQueue) UpdateRef(ctx context.Context, ref string, upt func(r *co
 	defer h.mu.Unlock()

 	var br controlapi.BuildHistoryRecord
-	if err := h.DB.View(func(tx *bolt.Tx) error {
+	if err := h.opt.DB.View(func(tx *bolt.Tx) error {
 		b := tx.Bucket([]byte(recordsBucket))
 		if b == nil {
 			return os.ErrNotExist
@@ -228,7 +378,7 @@ func (h *HistoryQueue) UpdateRef(ctx context.Context, ref string, upt func(r *co
 func (h *HistoryQueue) Status(ctx context.Context, ref string, st chan<- *client.SolveStatus) error {
 	h.init()
 	var br controlapi.BuildHistoryRecord
-	if err := h.DB.View(func(tx *bolt.Tx) error {
+	if err := h.opt.DB.View(func(tx *bolt.Tx) error {
 		b := tx.Bucket([]byte(recordsBucket))
 		if b == nil {
 			return os.ErrNotExist
@@ -250,7 +400,7 @@ func (h *HistoryQueue) Status(ctx context.Context, ref string, st chan<- *client
 		return nil
 	}

-	ra, err := h.ContentStore.ReaderAt(ctx, ocispecs.Descriptor{
+	ra, err := h.hContentStore.ReaderAt(ctx, ocispecs.Descriptor{
 		Digest:    br.Logs.Digest,
 		Size:      br.Logs.Size_,
 		MediaType: br.Logs.MediaType,
@@ -291,7 +441,7 @@ func (h *HistoryQueue) Status(ctx context.Context, ref string, st chan<- *client
 }

 func (h *HistoryQueue) update(ctx context.Context, rec controlapi.BuildHistoryRecord) error {
-	return h.DB.Update(func(tx *bolt.Tx) (err error) {
+	return h.opt.DB.Update(func(tx *bolt.Tx) (err error) {
 		b := tx.Bucket([]byte(recordsBucket))
 		if b == nil {
 			return nil
@@ -301,7 +451,7 @@ func (h *HistoryQueue) update(ctx context.Context, rec controlapi.BuildHistoryRe
 			return err
 		}

-		l, err := h.LeaseManager.Create(ctx, leases.WithID(h.leaseID(rec.Ref)))
+		l, err := h.hLeaseManager.Create(ctx, leases.WithID(h.leaseID(rec.Ref)))
 		created := true
 		if err != nil {
 			if !errors.Is(err, errdefs.ErrAlreadyExists) {
@@ -313,7 +463,7 @@ func (h *HistoryQueue) update(ctx context.Context, rec controlapi.BuildHistoryRe

 		defer func() {
 			if err != nil && created {
-				h.LeaseManager.Delete(ctx, l)
+				h.hLeaseManager.Delete(ctx, l)
 			}
 		}()

@@ -376,27 +526,27 @@ func (h *HistoryQueue) Delete(ctx context.Context, ref string) error {
 }

 func (h *HistoryQueue) OpenBlobWriter(ctx context.Context, mt string) (_ *Writer, err error) {
-	l, err := h.LeaseManager.Create(ctx, leases.WithRandomID(), leases.WithExpiration(5*time.Minute), leaseutil.MakeTemporary)
+	l, err := h.hLeaseManager.Create(ctx, leases.WithRandomID(), leases.WithExpiration(5*time.Minute), leaseutil.MakeTemporary)
 	if err != nil {
 		return nil, err
 	}

 	defer func() {
 		if err != nil {
-			h.LeaseManager.Delete(ctx, l)
+			h.hLeaseManager.Delete(ctx, l)
 		}
 	}()

 	ctx = leases.WithLease(ctx, l.ID)

-	w, err := content.OpenWriter(ctx, h.ContentStore, content.WithRef("history-"+h.leaseID(l.ID)))
+	w, err := content.OpenWriter(ctx, h.hContentStore, content.WithRef("history-"+h.leaseID(l.ID)))
 	if err != nil {
 		return nil, err
 	}

 	return &Writer{
 		mt:    mt,
-		lm:    h.LeaseManager,
+		lm:    h.hLeaseManager,
 		l:     l,
 		w:     w,
 		dgstr: digest.Canonical.Digester(),
@@ -577,7 +727,7 @@ func (h *HistoryQueue) Listen(ctx context.Context, req *controlapi.BuildHistoryR

 	if !req.ActiveOnly {
 		events := []*controlapi.BuildHistoryEvent{}
-		if err := h.DB.View(func(tx *bolt.Tx) error {
+		if err := h.opt.DB.View(func(tx *bolt.Tx) error {
 			b := tx.Bucket([]byte(recordsBucket))
 			if b == nil {
 				return nil
--- a/util/leaseutil/manager.go
+++ b/util/leaseutil/manager.go
@@ -35,7 +35,9 @@ func MakeTemporary(l *leases.Lease) error {
 	return nil
 }

-func WithNamespace(lm leases.Manager, ns string) leases.Manager {
+type Manager = nsLM
+
+func WithNamespace(lm leases.Manager, ns string) *Manager {
 	return &nsLM{manager: lm, ns: ns}
 }

@@ -44,6 +46,14 @@ type nsLM struct {
 	ns      string
 }

+func (l *nsLM) Namespace() string {
+	return l.ns
+}
+
+func (l *nsLM) WithNamespace(ns string) *Manager {
+	return WithNamespace(l.manager, ns)
+}
+
 func (l *nsLM) Create(ctx context.Context, opts ...leases.Opt) (leases.Lease, error) {
 	ctx = namespaces.WithNamespace(ctx, l.ns)
 	return l.manager.Create(ctx, opts...)
--- a/worker/base/worker.go
+++ b/worker/base/worker.go
@@ -11,7 +11,6 @@ import (
 	"github.com/containerd/containerd/diff"
 	"github.com/containerd/containerd/gc"
 	"github.com/containerd/containerd/images"
-	"github.com/containerd/containerd/leases"
 	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/remotes/docker"
 	"github.com/docker/docker/pkg/idtools"
@@ -30,6 +29,7 @@ import (
 	"github.com/moby/buildkit/identity"
 	"github.com/moby/buildkit/session"
 	"github.com/moby/buildkit/snapshot"
+	containerdsnapshot "github.com/moby/buildkit/snapshot/containerd"
 	"github.com/moby/buildkit/snapshot/imagerefchecker"
 	"github.com/moby/buildkit/solver"
 	"github.com/moby/buildkit/solver/llbsolver/mounts"
@@ -42,6 +42,7 @@ import (
 	"github.com/moby/buildkit/source/local"
 	"github.com/moby/buildkit/util/archutil"
 	"github.com/moby/buildkit/util/bklog"
+	"github.com/moby/buildkit/util/leaseutil"
 	"github.com/moby/buildkit/util/network"
 	"github.com/moby/buildkit/util/progress"
 	"github.com/moby/buildkit/util/progress/controller"
@@ -67,13 +68,13 @@ type WorkerOpt struct {
 	NetworkProviders map[pb.NetMode]network.Provider
 	Executor         executor.Executor
 	Snapshotter      snapshot.Snapshotter
-	ContentStore     content.Store
+	ContentStore     *containerdsnapshot.Store
 	Applier          diff.Applier
 	Differ           diff.Comparer
 	ImageStore       images.Store // optional
 	RegistryHosts    docker.RegistryHosts
 	IdentityMapping  *idtools.IdentityMapping
-	LeaseManager     leases.Manager
+	LeaseManager     *leaseutil.Manager
 	GarbageCollect   func(context.Context) (gc.Stats, error)
 	ParallelismSem   *semaphore.Weighted
 	MetadataStore    *metadata.Store
@@ -216,11 +217,11 @@ func (w *Worker) Close() error {
 	return rerr
 }

-func (w *Worker) ContentStore() content.Store {
+func (w *Worker) ContentStore() *containerdsnapshot.Store {
 	return w.WorkerOpt.ContentStore
 }

-func (w *Worker) LeaseManager() leases.Manager {
+func (w *Worker) LeaseManager() *leaseutil.Manager {
 	return w.WorkerOpt.LeaseManager
 }

--- a/worker/runc/runc.go
+++ b/worker/runc/runc.go
@@ -78,7 +78,7 @@ func NewWorkerOpt(root string, snFactory SnapshotterFactory, rootless bool, proc
 		return opt, err
 	}

-	c, err := local.NewStore(filepath.Join(root, "content"))
+	localstore, err := local.NewStore(filepath.Join(root, "content"))
 	if err != nil {
 		return opt, err
 	}
@@ -88,14 +88,14 @@ func NewWorkerOpt(root string, snFactory SnapshotterFactory, rootless bool, proc
 		return opt, err
 	}

-	mdb := ctdmetadata.NewDB(db, c, map[string]ctdsnapshot.Snapshotter{
+	mdb := ctdmetadata.NewDB(db, localstore, map[string]ctdsnapshot.Snapshotter{
 		snFactory.Name: s,
 	})
 	if err := mdb.Init(context.TODO()); err != nil {
 		return opt, err
 	}

-	c = containerdsnapshot.NewContentStore(mdb.ContentStore(), "buildkit")
+	c := containerdsnapshot.NewContentStore(mdb.ContentStore(), "buildkit")

 	id, err := base.ID(root)
 	if err != nil {
@@ -122,7 +122,6 @@ func NewWorkerOpt(root string, snFactory SnapshotterFactory, rootless bool, proc
 	}
 	lm := leaseutil.WithNamespace(ctdmetadata.NewLeaseManager(mdb), "buildkit")
 	snap := containerdsnapshot.NewSnapshotter(snFactory.Name, mdb.Snapshotter(snFactory.Name), "buildkit", idmap)
-
 	if err := cache.MigrateV2(
 		context.TODO(),
 		filepath.Join(root, "metadata.db"),
--- a/worker/worker.go
+++ b/worker/worker.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"io"

-	"github.com/containerd/containerd/content"
-	"github.com/containerd/containerd/leases"
 	"github.com/moby/buildkit/cache"
 	"github.com/moby/buildkit/client"
 	"github.com/moby/buildkit/client/llb"
@@ -13,7 +11,9 @@ import (
 	"github.com/moby/buildkit/exporter"
 	"github.com/moby/buildkit/frontend"
 	"github.com/moby/buildkit/session"
+	containerdsnapshot "github.com/moby/buildkit/snapshot/containerd"
 	"github.com/moby/buildkit/solver"
+	"github.com/moby/buildkit/util/leaseutil"
 	digest "github.com/opencontainers/go-digest"
 	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
 )
@@ -36,10 +36,10 @@ type Worker interface {
 	Prune(ctx context.Context, ch chan client.UsageInfo, opt ...client.PruneInfo) error
 	FromRemote(ctx context.Context, remote *solver.Remote) (cache.ImmutableRef, error)
 	PruneCacheMounts(ctx context.Context, ids []string) error
-	ContentStore() content.Store
+	ContentStore() *containerdsnapshot.Store
 	Executor() executor.Executor
 	CacheManager() cache.Manager
-	LeaseManager() leases.Manager
+	LeaseManager() *leaseutil.Manager
 }

 type Infos interface {