Merge pull request #7264 from janwaleed09/fix-contenthash-orphans

cache: fix content hash leak for removed records

cache/contenthash/checksum_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/containerd/containerd/v2/plugins/snapshots/native"
 	"github.com/moby/buildkit/cache"
 	"github.com/moby/buildkit/cache/metadata"
+	"github.com/moby/buildkit/client"
 	"github.com/moby/buildkit/session"
 	"github.com/moby/buildkit/snapshot"
 	containerdsnapshot "github.com/moby/buildkit/snapshot/containerd"
@@ -1436,6 +1437,69 @@ func TestPersistence(t *testing.T) {
 	require.Equal(t, dgstFileData0, dgst)
 }
 
+func TestNoPersistenceAfterRecordRemoval(t *testing.T) {
+	t.Parallel()
+	tmpdir := t.TempDir()
+
+	snapshotter, err := native.NewSnapshotter(filepath.Join(tmpdir, "snapshots"))
+	require.NoError(t, err)
+	cm, cleanup := setupCacheManager(t, tmpdir, "native", snapshotter)
+	t.Cleanup(cleanup)
+
+	ref := createRef(t, cm, []string{"ADD foo file data0"})
+	md := ensureOriginMetadata(ref)
+	require.NotEqual(t, ref.ID(), md.ID())
+
+	cc, err := newCacheContext(md)
+	require.NoError(t, err)
+	dgst, err := cc.Checksum(t.Context(), ref, "foo", ChecksumOpts{}, nil)
+	require.NoError(t, err)
+	require.Equal(t, dgstFileData0, dgst)
+
+	require.NoError(t, ref.Release(t.Context()))
+	require.NoError(t, cm.Prune(t.Context(), nil, client.PruneInfo{All: true}))
+
+	require.Error(t, cc.save())
+	_, err = md.GetExternal(keyContentHash)
+	require.Error(t, err)
+}
+
+func TestNoPersistenceAfterFinalize(t *testing.T) {
+	t.Parallel()
+	tmpdir := t.TempDir()
+
+	snapshotter, err := native.NewSnapshotter(filepath.Join(tmpdir, "snapshots"))
+	require.NoError(t, err)
+	cm, cleanup := setupCacheManager(t, tmpdir, "native", snapshotter)
+	t.Cleanup(cleanup)
+
+	ref := createRef(t, cm, []string{"ADD foo file data0"})
+	md := ensureOriginMetadata(ref)
+	require.NotEqual(t, ref.ID(), md.ID())
+
+	cc, err := newCacheContext(md)
+	require.NoError(t, err)
+	dgst, err := cc.Checksum(t.Context(), ref, "foo", ChecksumOpts{}, nil)
+	require.NoError(t, err)
+	require.Equal(t, dgstFileData0, dgst)
+	require.NoError(t, cc.save())
+	_, err = md.GetExternal(keyContentHash)
+	require.NoError(t, err)
+
+	mref, err := cm.New(t.Context(), ref, nil)
+	require.NoError(t, err)
+	require.NoError(t, mref.Release(t.Context()))
+	require.Eventually(t, func() bool {
+		_, err := md.GetExternal(keyContentHash)
+		return err != nil
+	}, 10*time.Second, 10*time.Millisecond)
+
+	require.Error(t, cc.save())
+	_, err = md.GetExternal(keyContentHash)
+	require.Error(t, err)
+	require.NoError(t, ref.Release(t.Context()))
+}
+
 func TestChecksumUpdateDirectory(t *testing.T) {
 	t.Parallel()
 	tmpdir := t.TempDir()

cache/manager.go

@@ -394,6 +394,11 @@ func (cm *cacheManager) init(ctx context.Context) error {
 			cm.LeaseManager.Delete(ctx, leases.Lease{ID: si.ID() + "-variants"})
 		}
 	}
+	if n, err := cm.MetadataStore.ClearOrphans(); err != nil {
+		bklog.G(ctx).Warnf("failed to remove external metadata of deleted cache records: %v", err)
+	} else if n > 0 {
+		bklog.G(ctx).Infof("removed external metadata of %d deleted cache records", n)
+	}
 	return nil
 }
 

cache/manager_test.go

@@ -3117,6 +3117,76 @@ func TestFinalizeCanceledContext(t *testing.T) {
 	require.NoError(t, iref.Release(ctx))
 }
 
+func TestStartupRemovesOrphanedExternalMetadata(t *testing.T) {
+	t.Parallel()
+
+	ctx := namespaces.WithNamespace(t.Context(), "buildkit-test")
+
+	tmpdir := t.TempDir()
+
+	snapshotter, err := native.NewSnapshotter(filepath.Join(tmpdir, "snapshots"))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, snapshotter.Close())
+	})
+
+	co, cleanup, err := newCacheManager(ctx, t, cmOpt{
+		tmpdir:          tmpdir,
+		snapshotter:     snapshotter,
+		snapshotterName: "native",
+	})
+	require.NoError(t, err)
+	t.Cleanup(cleanup)
+	cm := co.manager.(*cacheManager)
+
+	mref, err := cm.New(ctx, nil, nil, CachePolicyRetain)
+	require.NoError(t, err)
+	iref, err := mref.Commit(ctx)
+	require.NoError(t, err)
+	liveID := iref.ID()
+	require.NoError(t, iref.SetExternal("test", []byte("live")))
+	require.NoError(t, iref.Release(ctx))
+	require.NoError(t, mref.Release(ctx))
+
+	require.NoError(t, cm.MetadataStore.DB().Update(func(tx *bolt.Tx) error {
+		b, err := tx.CreateBucketIfNotExists([]byte("_external"))
+		if err != nil {
+			return err
+		}
+		b, err = b.CreateBucket([]byte("orphan"))
+		if err != nil {
+			return err
+		}
+		return b.Put([]byte("test"), []byte("leaked"))
+	}))
+	orphan, _ := cm.MetadataStore.Get("orphan")
+	dt, err := orphan.GetExternal("test")
+	require.NoError(t, err)
+	require.Equal(t, "leaked", string(dt))
+
+	require.NoError(t, cm.Close())
+	cleanup()
+
+	co, cleanup, err = newCacheManager(ctx, t, cmOpt{
+		tmpdir:          tmpdir,
+		snapshotter:     snapshotter,
+		snapshotterName: "native",
+	})
+	require.NoError(t, err)
+	t.Cleanup(cleanup)
+	cm = co.manager.(*cacheManager)
+
+	orphan, _ = cm.MetadataStore.Get("orphan")
+	_, err = orphan.GetExternal("test")
+	require.Error(t, err)
+
+	live, ok := cm.MetadataStore.Get(liveID)
+	require.True(t, ok)
+	dt, err = live.GetExternal("test")
+	require.NoError(t, err)
+	require.Equal(t, "live", string(dt))
+}
+
 func TestMountReadOnly(t *testing.T) {
 	t.Parallel()
 	if runtime.GOOS != "linux" {

cache/metadata/metadata.go

@@ -190,6 +190,40 @@ func (s *Store) Clear(id string) error {
 	}))
 }
 
+func (s *Store) ClearOrphans() (int, error) {
+	var n int
+	err := s.db.Update(func(tx *bolt.Tx) error {
+		external := tx.Bucket([]byte(externalBucket))
+		if external == nil {
+			return nil
+		}
+		var records *bolt.Cursor
+		if main := tx.Bucket([]byte(mainBucket)); main != nil {
+			records = main.Cursor()
+		}
+		var orphans [][]byte
+		if err := external.ForEach(func(k, _ []byte) error {
+			if records != nil {
+				if id, v := records.Seek(k); bytes.Equal(id, k) && v == nil {
+					return nil
+				}
+			}
+			orphans = append(orphans, append([]byte(nil), k...))
+			return nil
+		}); err != nil {
+			return err
+		}
+		for _, id := range orphans {
+			if err := external.DeleteBucket(id); err != nil {
+				return err
+			}
+		}
+		n = len(orphans)
+		return nil
+	})
+	return n, errors.WithStack(err)
+}
+
 func (s *Store) Update(id string, fn func(b *bolt.Bucket) error) error {
 	return errors.WithStack(s.db.Update(func(tx *bolt.Tx) error {
 		b, err := tx.CreateBucketIfNotExists([]byte(mainBucket))
@@ -330,6 +364,10 @@ func (s *StorageItem) GetExternal(k string) ([]byte, error) {
 
 func (s *StorageItem) SetExternal(k string, dt []byte) error {
 	return errors.WithStack(s.storage.db.Update(func(tx *bolt.Tx) error {
+		main := tx.Bucket([]byte(mainBucket))
+		if main == nil || main.Bucket([]byte(s.id)) == nil {
+			return errors.Wrapf(errNotFound, "record %s", s.id)
+		}
 		b, err := tx.CreateBucketIfNotExists([]byte(externalBucket))
 		if err != nil {
 			return errors.WithStack(err)

cache/metadata/metadata_test.go

@@ -187,8 +187,9 @@ func TestExternalData(t *testing.T) {
 	require.NoError(t, err)
 	defer s.Close()
 
+	require.NoError(t, s.Update("foo", func(*bolt.Bucket) error { return nil }))
 	si, ok := s.Get("foo")
-	require.False(t, ok)
+	require.True(t, ok)
 
 	err = si.SetExternal("ext1", []byte("data"))
 	require.NoError(t, err)
@@ -214,6 +215,52 @@ func TestExternalData(t *testing.T) {
 	si, _ = s.Get("foo")
 	_, err = si.GetExternal("ext1")
 	require.Error(t, err)
+
+	require.ErrorIs(t, si.SetExternal("ext1", []byte("late")), errNotFound)
+	_, err = si.GetExternal("ext1")
+	require.Error(t, err)
+}
+
+func TestClearOrphans(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "storage.db")
+
+	s, err := NewStore(dbPath)
+	require.NoError(t, err)
+
+	require.NoError(t, s.Update("live", func(*bolt.Bucket) error { return nil }))
+	si, _ := s.Get("live")
+	require.NoError(t, si.SetExternal("ext1", []byte("data")))
+
+	require.NoError(t, s.db.Update(func(tx *bolt.Tx) error {
+		b, err := tx.CreateBucketIfNotExists([]byte(externalBucket))
+		if err != nil {
+			return err
+		}
+		b, err = b.CreateBucket([]byte("gone"))
+		if err != nil {
+			return err
+		}
+		return b.Put([]byte("ext1"), []byte("leaked"))
+	}))
+
+	n, err := s.ClearOrphans()
+	require.NoError(t, err)
+	require.Equal(t, 1, n)
+
+	dt, err := si.GetExternal("ext1")
+	require.NoError(t, err)
+	require.Equal(t, "data", string(dt))
+
+	gone, _ := s.Get("gone")
+	_, err = gone.GetExternal("ext1")
+	require.Error(t, err)
+
+	n, err = s.ClearOrphans()
+	require.NoError(t, err)
+	require.Equal(t, 0, n)
+	require.NoError(t, s.Close())
 }
 
 func TestIndexReplacement(t *testing.T) {