workflow: use heartbeats in acquire-pipeline

It is cheaper to retry within the activity and let it run as long as it can
heartbeat. It reduces the pressure put in the workflow engine.

internal/workflow/activities/acquire_pipeline.go

@@ -2,10 +2,14 @@ package activities
 
 import (
 	"context"
-	"errors"
+	"fmt"
+	"time"
+
+	"go.uber.org/cadence/activity"
 
 	wferrors "github.com/artefactual-labs/enduro/internal/workflow/errors"
 	"github.com/artefactual-labs/enduro/internal/workflow/manager"
+	"github.com/cenkalti/backoff/v3"
 )
 
 // AcquirePipelineActivity acquires a lock in the weighted semaphore associated
@@ -24,9 +28,22 @@ func (a *AcquirePipelineActivity) Execute(ctx context.Context, pipelineName stri
 		return wferrors.NonRetryableError(err)
 	}
 
-	if ok := p.TryAcquire(); !ok {
-		return errors.New("error acquiring pipeline")
-	}
+	var errAcquirePipeline = fmt.Errorf("error acquring semaphore: busy")
+
+	err = backoff.RetryNotify(
+		func() (err error) {
+			ok := p.TryAcquire()
+			if !ok {
+				err = errAcquirePipeline
+			}
+
+			return err
+		},
+		backoff.WithContext(backoff.NewConstantBackOff(time.Second*5), ctx),
+		func(err error, duration time.Duration) {
+			activity.RecordHeartbeat(ctx)
+		},
+	)
 
-	return nil
+	return err
 }

internal/workflow/activities/activities.go

@@ -1,3 +1,4 @@
+// Package activities implements Enduro's workflow activities.
 package activities
 
 const (

internal/workflow/semaphore.go

@@ -11,7 +11,6 @@ import (
 	"github.com/artefactual-labs/enduro/internal/workflow/manager"
 
 	"github.com/go-logr/logr"
-	"go.uber.org/cadence"
 	"go.uber.org/cadence/workflow"
 )
 
@@ -20,16 +19,9 @@ func acquirePipeline(ctx workflow.Context, manager *manager.Manager, pipelineNam
 	{
 		ctx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 			ScheduleToStartTimeout: forever,
-			StartToCloseTimeout:    time.Second * 5,
-			WaitForCancellation:    true,
-			RetryPolicy: &cadence.RetryPolicy{
-				InitialInterval:    time.Second * 2,
-				BackoffCoefficient: 1,
-				MaximumInterval:    time.Second * 2,
-				ExpirationInterval: forever,
-			},
+			StartToCloseTimeout:    forever,
+			HeartbeatTimeout:       time.Minute,
 		})
-
 		if err := workflow.ExecuteActivity(ctx, activities.AcquirePipelineActivityName, pipelineName).Get(ctx, nil); err != nil {
 			return fmt.Errorf("error acquiring pipeline: %w", err)
 		}