fix gc after delete history records

Current implementation based on leases.SynchronousDelete only works
with the containerd worker and is ignored otherwise. This means that
blobs referenced by history records were left on disc until the
periodic background GC was initialized later.

Signed-off-by: Tonis Tiigi <[email protected]>

Author: tonistiigi

cmd/buildkitd/main.go

@@ -850,6 +850,7 @@ func newController(c *cli.Context, cfg *config.Config) (*control.Controller, err
 		LeaseManager:              w.LeaseManager(),
 		ContentStore:              w.ContentStore(),
 		HistoryConfig:             cfg.History,
+		GarbageCollect:            w.GarbageCollect,
 	})
 }
 

control/control.go

@@ -70,6 +70,7 @@ type Opt struct {
 	LeaseManager              *leaseutil.Manager
 	ContentStore              *containerdsnapshot.Store
 	HistoryConfig             *config.HistoryConfig
+	GarbageCollect            func(context.Context) error
 }
 
 type Controller struct { // TODO: ControlService
@@ -89,10 +90,11 @@ func NewController(opt Opt) (*Controller, error) {
 	gatewayForwarder := controlgateway.NewGatewayForwarder()
 
 	hq, err := llbsolver.NewHistoryQueue(llbsolver.HistoryQueueOpt{
-		DB:           opt.HistoryDB,
-		LeaseManager: opt.LeaseManager,
-		ContentStore: opt.ContentStore,
-		CleanConfig:  opt.HistoryConfig,
+		DB:             opt.HistoryDB,
+		LeaseManager:   opt.LeaseManager,
+		ContentStore:   opt.ContentStore,
+		CleanConfig:    opt.HistoryConfig,
+		GarbageCollect: opt.GarbageCollect,
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create history queue")

solver/llbsolver/history.go

@@ -42,10 +42,11 @@ const (
 )
 
 type HistoryQueueOpt struct {
-	DB           db.Transactor
-	LeaseManager *leaseutil.Manager
-	ContentStore *containerdsnapshot.Store
-	CleanConfig  *config.HistoryConfig
+	DB             db.Transactor
+	LeaseManager   *leaseutil.Manager
+	ContentStore   *containerdsnapshot.Store
+	CleanConfig    *config.HistoryConfig
+	GarbageCollect func(context.Context) error
 }
 
 type HistoryQueue struct {
@@ -323,7 +324,7 @@ func (h *HistoryQueue) gc() error {
 	now := time.Now()
 	for _, r := range records[h.opt.CleanConfig.MaxEntries:] {
 		if now.Add(-h.opt.CleanConfig.MaxAge.Duration).After(r.CompletedAt.AsTime()) {
-			if err := h.delete(r.Ref, false); err != nil {
+			if _, err := h.delete(r.Ref); err != nil {
 				return err
 			}
 		}
@@ -365,18 +366,18 @@ func (h *HistoryQueue) clearOrphans() error {
 
 	for _, r := range records {
 		bklog.G(ctx).Warnf("deleting build record %s due to missing blobs", r.Ref)
-		if err := h.delete(r.Ref, false); err != nil {
+		if _, err := h.delete(r.Ref); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-func (h *HistoryQueue) delete(ref string, sync bool) error {
+func (h *HistoryQueue) delete(ref string) (bool, error) {
 	if _, ok := h.refs[ref]; ok {
 		h.deleted[ref] = struct{}{}
-		return nil
+		return false, nil
 	}
 	delete(h.deleted, ref)
 	h.ps.Send(&controlapi.BuildHistoryEvent{
@@ -389,19 +390,15 @@ func (h *HistoryQueue) delete(ref string, sync bool) error {
 			return errors.Wrapf(os.ErrNotExist, "failed to retrieve bucket %s", recordsBucket)
 		}
 		err1 := b.Delete([]byte(ref))
-		var opts []leases.DeleteOpt
-		if sync {
-			opts = append(opts, leases.SynchronousDelete)
-		}
-		err2 := h.hLeaseManager.Delete(context.TODO(), leases.Lease{ID: h.leaseID(ref)}, opts...)
+		err2 := h.hLeaseManager.Delete(context.TODO(), leases.Lease{ID: h.leaseID(ref)})
 		if err1 != nil {
 			return err1
 		}
 		return err2
 	}); err != nil {
-		return err
+		return false, err
 	}
-	return nil
+	return true, nil
 }
 
 func (h *HistoryQueue) init() error {
@@ -683,7 +680,14 @@ func (h *HistoryQueue) Delete(ctx context.Context, ref string) error {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	return h.delete(ref, true)
+	v, err := h.delete(ref)
+	if err != nil {
+		return err
+	}
+	if v {
+		return h.opt.GarbageCollect(ctx)
+	}
+	return nil
 }
 
 func (h *HistoryQueue) OpenBlobWriter(ctx context.Context, mt string) (_ *Writer, err error) {
@@ -909,7 +913,7 @@ func (h *HistoryQueue) Listen(ctx context.Context, req *controlapi.BuildHistoryR
 			if _, ok := h.deleted[req.Ref]; ok {
 				if h.refs[req.Ref] == 0 {
 					delete(h.refs, req.Ref)
-					h.delete(req.Ref, false)
+					h.delete(req.Ref)
 				}
 			}
 			h.mu.Unlock()

worker/base/worker.go

@@ -211,6 +211,14 @@ func NewWorker(ctx context.Context, opt WorkerOpt) (*Worker, error) {
 	}, nil
 }
 
+func (w *Worker) GarbageCollect(ctx context.Context) error {
+	if w.WorkerOpt.GarbageCollect == nil {
+		return nil
+	}
+	_, err := w.WorkerOpt.GarbageCollect(ctx)
+	return err
+}
+
 func (w *Worker) Close() error {
 	var rerr error
 	if err := w.MetadataStore.Close(); err != nil {

worker/worker.go

@@ -40,6 +40,7 @@ type Worker interface {
 	Executor() executor.Executor
 	CacheManager() cache.Manager
 	LeaseManager() *leaseutil.Manager
+	GarbageCollect(context.Context) error
 }
 
 type Infos interface {