Add option to reject duplicates

Fixes #545.

Co-authored-by: Jesús García Crespo <[email protected]>

enduro.toml

@@ -29,6 +29,7 @@ bucket = "sips"
 pipeline = "am"
 retentionPeriod = "10s"
 stripTopLevelDir = true
+rejectDuplicates = false
 
 [[watcher.filesystem]]
 name = "dev-fs"
@@ -38,6 +39,7 @@ pipeline = "am"
 completedDir = "./hack/landfill"
 ignore = '(^\.gitkeep)|(^*\.mft)$'
 stripTopLevelDir = true
+rejectDuplicates = false
 
 [[pipeline]]
 name = "am"

internal/api/design/batch.go

@@ -17,6 +17,7 @@ var _ = Service("batch", func() {
 			Attribute("processing_config", String)
 			Attribute("completed_dir", String)
 			Attribute("retention_period", String)
+			Attribute("reject_duplicates", Boolean, func() { Default(false) })
 			Required("path")
 		})
 		Result(BatchResult)

internal/batch/service.go

@@ -71,6 +71,7 @@ func (s *batchImpl) Submit(ctx context.Context, payload *goabatch.SubmitPayload)
 		}
 		input.RetentionPeriod = &dur
 	}
+	input.RejectDuplicates = payload.RejectDuplicates
 	opts := temporalsdk_client.StartWorkflowOptions{
 		ID:                       BatchWorkflowID,
 		WorkflowIDReusePolicy:    temporalapi_enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,

internal/batch/workflow.go

@@ -28,6 +28,7 @@ type BatchWorkflowInput struct {
 	ProcessingConfig string
 	CompletedDir     string
 	RetentionPeriod  *time.Duration
+	RejectDuplicates bool
 }
 
 func BatchWorkflow(ctx temporalsdk_workflow.Context, params BatchWorkflowInput) error {
@@ -69,6 +70,7 @@ func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput)
 			ProcessingConfig: params.ProcessingConfig,
 			CompletedDir:     params.CompletedDir,
 			RetentionPeriod:  params.RetentionPeriod,
+			RejectDuplicates: params.RejectDuplicates,
 		}
 		_ = a.batchsvc.InitProcessingWorkflow(ctx, &req)
 	}

internal/collection/collection.go

@@ -18,6 +18,7 @@ type Service interface {
 	// Goa returns an implementation of the goacollection Service.
 	Goa() goacollection.Service
 	Create(context.Context, *Collection) error
+	CheckDuplicate(ctx context.Context, id uint) (bool, error)
 	UpdateWorkflowStatus(ctx context.Context, ID uint, name string, workflowID, runID, transferID, aipID, pipelineID string, status Status, storedAt time.Time) error
 	SetStatus(ctx context.Context, ID uint, status Status) error
 	SetStatusInProgress(ctx context.Context, ID uint, startedAt time.Time) error
@@ -88,6 +89,16 @@ func (svc *collectionImpl) Create(ctx context.Context, col *Collection) error {
 	return nil
 }
 
+func (svc *collectionImpl) CheckDuplicate(ctx context.Context, id uint) (bool, error) {
+	query := `SELECT EXISTS(SELECT 1 FROM collection c1 WHERE c1.name = (SELECT name FROM collection WHERE id = ?) AND c1.id <> ? AND c1.status NOT IN (3, 6))`
+	var exists bool
+	err := svc.db.GetContext(ctx, &exists, query, id, id)
+	if err != nil {
+		return false, fmt.Errorf("sql error: %w", err)
+	}
+	return exists, nil
+}
+
 func publishEvent(ctx context.Context, events EventService, eventType string, id uint) {
 	// TODO: publish updated collection?
 	var item *goacollection.EnduroStoredCollection

internal/collection/workflow.go

@@ -55,6 +55,9 @@ type ProcessingWorkflowRequest struct {
 
 	// Processing configuration name.
 	ProcessingConfig string
+
+	// Whether we reject duplicates based on name (key).
+	RejectDuplicates bool
 }
 
 func InitProcessingWorkflow(ctx context.Context, c temporalsdk_client.Client, taskQueue string, req *ProcessingWorkflowRequest) error {

internal/watcher/config.go

@@ -37,6 +37,7 @@ type FilesystemConfig struct {
 	RetentionPeriod  *time.Duration
 	CompletedDir     string
 	StripTopLevelDir bool
+	RejectDuplicates bool
 }
 
 // See minio.go for more.
@@ -56,4 +57,5 @@ type MinioConfig struct {
 	Pipeline         []string
 	RetentionPeriod  *time.Duration
 	StripTopLevelDir bool
+	RejectDuplicates bool
 }

internal/watcher/event.go

@@ -29,6 +29,9 @@ type BlobEvent struct {
 	// Whether the top-level directory is meant to be stripped.
 	StripTopLevelDir bool
 
+	// Whether duplicates are rejected or not.
+	RejectDuplicates bool
+
 	// Key of the blob.
 	Key string
 
@@ -46,6 +49,7 @@ func NewBlobEvent(w Watcher, key string, isDir bool) *BlobEvent {
 		RetentionPeriod:  w.RetentionPeriod(),
 		CompletedDir:     w.CompletedDir(),
 		StripTopLevelDir: w.StripTopLevelDir(),
+		RejectDuplicates: w.RejectDuplicates(),
 		Key:              key,
 		IsDir:            isDir,
 	}

internal/watcher/filesystem.go

@@ -76,6 +76,7 @@ func NewFilesystemWatcher(ctx context.Context, config *FilesystemConfig) (*files
 			retentionPeriod:  config.RetentionPeriod,
 			completedDir:     config.CompletedDir,
 			stripTopLevelDir: config.StripTopLevelDir,
+			rejectDuplicates: config.RejectDuplicates,
 		},
 	}
 

internal/watcher/minio.go

@@ -70,6 +70,7 @@ func NewMinioWatcher(ctx context.Context, config *MinioConfig) (*minioWatcher, e
 			pipeline:         config.Pipeline,
 			retentionPeriod:  config.RetentionPeriod,
 			stripTopLevelDir: config.StripTopLevelDir,
+			rejectDuplicates: config.RejectDuplicates,
 		},
 	}, nil
 }

internal/watcher/watcher.go

@@ -29,6 +29,7 @@ type Watcher interface {
 	RetentionPeriod() *time.Duration
 	CompletedDir() string
 	StripTopLevelDir() bool
+	RejectDuplicates() bool
 
 	// Full path of the watched bucket when available, empty string otherwise.
 	Path() string
@@ -42,6 +43,7 @@ type commonWatcherImpl struct {
 	retentionPeriod  *time.Duration
 	completedDir     string
 	stripTopLevelDir bool
+	rejectDuplicates bool
 }
 
 func (w *commonWatcherImpl) String() string {
@@ -64,6 +66,10 @@ func (w *commonWatcherImpl) StripTopLevelDir() bool {
 	return w.stripTopLevelDir
 }
 
+func (w *commonWatcherImpl) RejectDuplicates() bool {
+	return w.rejectDuplicates
+}
+
 type Service interface {
 	// Watchers return all known watchers.
 	Watchers() []Watcher

internal/workflow/local_activities.go

@@ -60,6 +60,10 @@ func updatePackageLocalActivity(ctx context.Context, logger logr.Logger, colsvc
 	return nil
 }
 
+func checkDuplicatePackageLocalActivity(ctx context.Context, logger logr.Logger, colsvc collection.Service, id uint) (bool, error) {
+	return colsvc.CheckDuplicate(ctx, id)
+}
+
 func loadConfigLocalActivity(ctx context.Context, m *manager.Manager, pipeline string, tinfo *TransferInfo) (*TransferInfo, error) {
 	p, err := m.Pipelines.ByName(pipeline)
 	if err != nil {

internal/workflow/processing.go

@@ -226,6 +226,21 @@ func (w *ProcessingWorkflow) Execute(ctx temporalsdk_workflow.Context, req *coll
 		}).Get(activityOpts, nil)
 	}()
 
+	// Reject duplicate collection if applicable.
+	{
+		if req.RejectDuplicates {
+			var exists bool
+			activityOpts := withLocalActivityOpts(ctx)
+			err := temporalsdk_workflow.ExecuteLocalActivity(activityOpts, checkDuplicatePackageLocalActivity, w.manager.Logger, w.manager.Collection, tinfo.CollectionID).Get(activityOpts, &exists)
+			if err != nil {
+				return fmt.Errorf("error checking duplicate: %v", err)
+			}
+			if exists {
+				return fmt.Errorf("duplicate detected: key: %s", tinfo.Key)
+			}
+		}
+	}
+
 	// Extract details from transfer name.
 	{
 		activityOpts := withLocalActivityWithoutRetriesOpts(ctx)