fix(storage): bound reconciliation and cleanup memory usage

Author: Mmx233

internal/cache/service.go

@@ -43,7 +43,7 @@ const (
 )
 
 // MaxBlockListEntries is the Azure block blob protocol limit.
-const MaxBlockListEntries = 50_000
+const MaxBlockListEntries = storage.MaxIndexedObjects
 
 var (
 	ErrNoWriteScope        = errors.New("no scope with write permission found")
@@ -452,13 +452,17 @@ func (s *Service) CompleteUpload(ctx context.Context, key, version string, scope
 				currentUpload.FinishedPartUploadCount,
 			)
 		}
-		parts, inspectErr := s.storage.InspectFolder(activityCtx, partsFolderName(currentUpload.FolderName))
-		if inspectErr != nil {
-			return fmt.Errorf("inspect finalized cache parts: %w", inspectErr)
-		}
-		sizeBytes, validationErr = parts.LogicalIndexedSize(partCount)
+		sizeBytes, validationErr = s.storage.InspectIndexedFolder(
+			activityCtx,
+			partsFolderName(currentUpload.FolderName),
+			partCount,
+		)
 		if validationErr != nil {
-			return fmt.Errorf("%w: %v", ErrPartCountMismatch, validationErr)
+			if errors.Is(validationErr, storage.ErrIndexedObjectMissing) ||
+				errors.Is(validationErr, storage.ErrIndexedObjectLimitExceeded) {
+				return fmt.Errorf("%w: %v", ErrPartCountMismatch, validationErr)
+			}
+			return fmt.Errorf("inspect finalized cache parts: %w", validationErr)
 		}
 		return nil
 	})

internal/cache/service_test.go

@@ -222,6 +222,45 @@ func TestCompleteUploadKeepsLegacyUploadAfterTransientStorageError(t *testing.T)
 	require.NoError(t, err)
 }
 
+func TestCompleteUploadKeepsUploadAfterTransientIndexedInspectionError(t *testing.T) {
+	ctx, client, filesystem := newTestServiceDeps(t)
+	adapter := &storageCallTrackingAdapter{Adapter: filesystem, inspectIndexedErr: errInjectedStorageFailure}
+	service := NewService(Options{DB: client, Storage: adapter})
+	scope := writableScope()
+
+	upload, err := service.CreateUpload(ctx, "key", "version", scope)
+	require.NoError(t, err)
+	require.NoError(t, service.UploadPart(ctx, upload.UploadID, bytes.NewBufferString("data")))
+
+	_, err = service.CompleteUpload(ctx, "key", "version", scope)
+	require.ErrorIs(t, err, errInjectedStorageFailure)
+	_, err = client.Upload.Get(ctx, upload.UploadID)
+	require.NoError(t, err)
+}
+
+func TestCompleteUploadRejectsAndDeletesOverLimitRestoredSession(t *testing.T) {
+	ctx, client, filesystem := newTestServiceDeps(t)
+	service := NewService(Options{DB: client, Storage: filesystem})
+	scope := writableScope()
+	const uploadID = int64(42)
+	client.Upload.Create().
+		SetID(uploadID).
+		SetKey("restored-key").
+		SetVersion("version").
+		SetScope(scope.Scopes[0].Scope).
+		SetRepoId(scope.RepoID).
+		SetCreatedAt(time.Now().UnixMilli()).
+		SetFinishedPartUploadCount(storage.MaxIndexedObjects + 1).
+		SetCommittedPartCount(storage.MaxIndexedObjects + 1).
+		SetFolderName("restored-upload").
+		SaveX(ctx)
+
+	_, err := service.CompleteUpload(ctx, "restored-key", "version", scope)
+	require.ErrorIs(t, err, ErrPartCountMismatch)
+	_, err = client.Upload.Get(ctx, uploadID)
+	require.True(t, ent.IsNotFound(err))
+}
+
 func TestCompleteUploadRejectsBlocksWithoutCommittedBlockList(t *testing.T) {
 	ctx, client, filesystem := newTestServiceDeps(t)
 	service := NewService(Options{DB: client, Storage: filesystem})
@@ -1287,6 +1326,7 @@ type storageCallTrackingAdapter struct {
 	downloadCalls     int
 	objectExistsCalls []string
 	countErr          error
+	inspectIndexedErr error
 }
 
 func (s *storageCallTrackingAdapter) CountFilesInFolder(ctx context.Context, folderName string) (int, error) {
@@ -1302,6 +1342,13 @@ func (s *storageCallTrackingAdapter) CreateDownloadStream(ctx context.Context, o
 	return s.Adapter.CreateDownloadStream(ctx, objectName)
 }
 
+func (s *storageCallTrackingAdapter) InspectIndexedFolder(ctx context.Context, folderName string, expectedObjects int) (int64, error) {
+	if s.inspectIndexedErr != nil {
+		return 0, s.inspectIndexedErr
+	}
+	return s.Adapter.InspectIndexedFolder(ctx, folderName, expectedObjects)
+}
+
 func (s *storageCallTrackingAdapter) ObjectExists(ctx context.Context, objectName string) (bool, error) {
 	s.objectExistsCalls = append(s.objectExistsCalls, objectName)
 	return s.Adapter.ObjectExists(ctx, objectName)

internal/cache/upload_activity_test.go

@@ -331,10 +331,10 @@ type afterInspectStorage struct {
 	once  sync.Once
 }
 
-func (s *afterInspectStorage) InspectFolder(ctx context.Context, folderName string) (storage.FolderContents, error) {
-	contents, err := s.Adapter.InspectFolder(ctx, folderName)
+func (s *afterInspectStorage) InspectIndexedFolder(ctx context.Context, folderName string, expectedObjects int) (int64, error) {
+	sizeBytes, err := s.Adapter.InspectIndexedFolder(ctx, folderName, expectedObjects)
 	if err == nil {
 		s.once.Do(s.after)
 	}
-	return contents, err
+	return sizeBytes, err
 }