api/controller/cleanup_upload_use.go

Rama main · réplica de lectura del repositorio original.

package controller

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"

	"github.com/goodrain/rainbond/db"
	"github.com/goodrain/rainbond/db/model"
	guard "github.com/goodrain/rainbond/pkg/cleanup"
	"github.com/jinzhu/gorm"
)

// admitUploadUse persists a producer before native storage work. Upload content
// may introduce image references during later parsing, so its unresolved scope
// follows the existing native builder's conservative all-store admission.
func admitUploadUse(database *gorm.DB, eventID, sessionID, action string) (func(bool) error, error) {
	if database == nil || !uploadInventoryEventID.MatchString(eventID) || !uploadInventoryEventID.MatchString(sessionID) {
		return nil, guard.ErrCoordinationChanged
	}
	if action != "chunk" && action != "complete" && action != "cancel" && action != "expire" {
		return nil, guard.ErrCoordinationChanged
	}
	stores, err := guard.DiscoverStores(database)
	if err != nil {
		return nil, err
	}
	id, err := guard.NewActivationRevision()
	if err != nil {
		return nil, err
	}
	fingerprint := sha256.Sum256([]byte(eventID + "\x00" + sessionID + "\x00" + action))
	requests := []guard.CoordinationRequest{}
	finish := func(confirmed bool) error {
		var first error
		for _, request := range requests {
			if err := guard.FinishOperation(database, request, confirmed); err != nil && first == nil {
				first = err
			}
		}
		return first
	}
	for _, store := range stores {
		identity := sha256.Sum256([]byte(id + "\x00" + store.StorageID))
		request := guard.CoordinationRequest{StorageID: store.StorageID, Generation: store.Generation, OperationID: hex.EncodeToString(identity[:]), Owner: "api-package-" + action, Kind: "producer", Scope: "*", Fingerprint: hex.EncodeToString(fingerprint[:])}
		created, err := guard.AcquireOperation(database, request)
		if err != nil || !created {
			_ = finish(true) // No storage work started; release only this call's acknowledged admissions.
			if err != nil {
				return nil, err
			}
			return nil, guard.ErrCoordinationChanged
		}
		requests = append(requests, request)
	}
	sessionRequest := guard.PackageUploadRequest{OperationID: id, SessionID: sessionID, EventID: eventID, Action: action}
	if _, err := guard.AcquirePackageUploadRequest(database, sessionRequest); err != nil {
		// Only explicit pre-admission rejection proves no session grant was committed.
		rejected := errors.Is(err, guard.ErrCoordinationBusy) || errors.Is(err, guard.ErrCoordinationChanged) || errors.Is(err, guard.ErrCoordinationUncertain)
		_ = finish(rejected)
		return nil, err
	}
	return func(confirmed bool) error {
		if err := guard.FinishPackageUploadRequest(database, sessionRequest, confirmed); err != nil {
			_ = finish(false)
			return err
		}
		return finish(confirmed)
	}, nil
}

func (m *ChunkUploadManager) admitSessionUse(session *model.UploadSession, action string) (func(bool) error, error) {
	if m.admitUse != nil {
		return m.admitUse(session, action)
	}
	return admitUploadUse(db.GetManager().DB(), session.EventID, session.ID, action)
}

func (m *ChunkUploadManager) expireSessionChunks(session *model.UploadSession) (resultErr error) {
	finish, err := m.admitSessionUse(session, "expire")
	if err != nil {
		return err
	}
	confirmed := false
	defer func() {
		if err := finish(confirmed); resultErr == nil {
			resultErr = err
		}
	}()
	if err := m.cleanupSessionChunks(session.ID); err != nil {
		return err
	}
	confirmed = true
	return nil
}