Track buckets in DB instead of in-memory

.gitignore

@@ -24,6 +24,7 @@ peer.key
 wallet/
 duplicateGuard/
 pinQueue/
+stagingdata/
 
 cidlistsdir/*
 

cmd/estuary-shuttle/rpc.go

@@ -321,7 +321,7 @@ func (d *Shuttle) handleRpcTakeContent(ctx context.Context, cmd *drpc.TakeConten
 }
 
 func (s *Shuttle) handleRpcAggregateStagedContent(ctx context.Context, aggregate *drpc.AggregateContents) error {
-	// only progress if aggr is not allready in progress
+	// only progress if aggr is not already in progress
 	if !s.markStartAggr(aggregate.DBID) {
 		return nil
 	}

contentmgr/gc.go

@@ -3,7 +3,6 @@ package contentmgr
 import (
 	"context"
 	"fmt"
-
 	"github.com/application-research/estuary/util"
 	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
@@ -152,33 +151,46 @@ func (cm *ContentManager) UnpinContent(ctx context.Context, contid uint) error {
 		return err
 	}
 
-	if err := cm.DB.Delete(&util.Content{ID: pin.ID}).Error; err != nil {
-		return err
-	}
-
+	// delete object refs rows for deleted content
 	if err := cm.DB.Where("content = ?", pin.ID).Delete(&util.ObjRef{}).Error; err != nil {
 		return err
 	}
 
+	// delete objects that no longer have obj refs
 	if err := cm.clearUnreferencedObjects(ctx, objs); err != nil {
 		return err
 	}
 
+	// delete objects from blockstore that are no longer present in db
 	for _, o := range objs {
 		// TODO: this is safe, but... slow?
 		if _, err := cm.deleteIfNotPinned(ctx, o); err != nil {
 			return err
 		}
 	}
 
-	buckets, ok := cm.Buckets[pin.UserID]
-	if ok {
-		for _, bucket := range buckets {
-			if _, err := cm.tryRemoveContent(bucket, pin); err != nil {
+	// delete from contents table and adjust aggregate size, if applicable, in one tx
+	err = cm.DB.Transaction(func(tx *gorm.DB) error {
+		// delete contid row from contents table
+		if err := tx.Model(util.Content{}).Delete(&util.Content{ID: pin.ID}).Error; err != nil {
+			return err
+		}
+
+		if pin.AggregatedIn > 0 {
+			// decrease aggregate's size by cont's size in contents table
+			if err := tx.Model(util.Content{}).
+				Where("id = ?", pin.AggregatedIn).
+				UpdateColumn("size", gorm.Expr("size - ?", pin.Size)).Error; err != nil {
 				return err
 			}
 		}
+
+		return nil
+	})
+	if err != nil {
+		return err
 	}
+
 	return nil
 }
 

contentmgr/pinning.go

@@ -336,10 +336,8 @@ func (cm *ContentManager) selectLocationForRetrieval(ctx context.Context, cont u
 }
 
 func (cm *ContentManager) primaryStagingLocation(ctx context.Context, uid uint) string {
-	cm.BucketLk.Lock()
-	defer cm.BucketLk.Unlock()
-	zones, ok := cm.Buckets[uid]
-	if !ok {
+	var zones []util.Content
+	if err := cm.DB.Find(&zones, "user_id = ? and aggregate", uid).Error; err != nil {
 		return ""
 	}
 
@@ -428,6 +426,10 @@ func (cm *ContentManager) handlePinningComplete(ctx context.Context, handle stri
 		}).Error; err != nil {
 			return xerrors.Errorf("failed to update content in database: %w", err)
 		}
+
+		// if the content is a consolidated aggregate, it means aggregation has been completed and we can mark as finished
+		cm.MarkFinishedAggregating(cont)
+
 		// after aggregate is done, make deal for it
 		cm.ToCheck(cont.ID)
 		return nil