diff --git a/cache/contenthash/checksum_test.go b/cache/contenthash/checksum_test.go index 7a2e2e374..ac5611f05 100644 --- a/cache/contenthash/checksum_test.go +++ b/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() diff --git a/cache/manager.go b/cache/manager.go index 57403ead5..07e2e8ad8 100644 --- a/cache/manager.go +++ b/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 } diff --git a/cache/manager_test.go b/cache/manager_test.go index 629a4edf1..667c97a8b 100644 --- a/cache/manager_test.go +++ b/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" { diff --git a/cache/metadata/metadata.go b/cache/metadata/metadata.go index 01c6608e7..fd90c4111 100644 --- a/cache/metadata/metadata.go +++ b/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) diff --git a/cache/metadata/metadata_test.go b/cache/metadata/metadata_test.go index c8556dde7..13375a578 100644 --- a/cache/metadata/metadata_test.go +++ b/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) {