feat(polling): remove polling for waiting stages

orca-core/src/main/groovy/com/netflix/spinnaker/orca/pipeline/RestrictExecutionDuringTimeWindow.groovy

@@ -16,6 +16,7 @@
 
 package com.netflix.spinnaker.orca.pipeline
 
+import java.time.Duration
 import java.util.concurrent.TimeUnit
 import com.google.common.annotations.VisibleForTesting
 import com.netflix.spinnaker.orca.ExecutionStatus
@@ -60,8 +61,21 @@ class RestrictExecutionDuringTimeWindow implements StageDefinitionBuilder {
     @Value('${tasks.executionWindow.timezone:America/Los_Angeles}')
     String timeZoneId
 
+    @Override
+    long getDynamicBackoffPeriod(Duration taskDuration) {
+      if (taskDuration < Duration.ofMillis(timeout)) {
+        // wait until timeout is over to poll
+        return Duration.ofMillis(timeout).toMillis()
+      } else {
+        //start polling normally after timeout to account for delays like throttling
+        return backoffPeriod
+      }
+    }
+
     @Override
     TaskResult execute(Stage stage) {
+      stage.getTopLevelTimeout().ifPresent({ timeout = it })
+
       Date now = new Date()
       Date scheduledTime
       try {

orca-core/src/main/groovy/com/netflix/spinnaker/orca/pipeline/tasks/WaitTask.groovy

@@ -16,6 +16,7 @@
 
 package com.netflix.spinnaker.orca.pipeline.tasks
 
+import java.time.Duration
 import java.util.concurrent.TimeUnit
 import com.netflix.spinnaker.orca.RetryableTask
 import com.netflix.spinnaker.orca.TaskResult
@@ -30,14 +31,27 @@ import static com.netflix.spinnaker.orca.ExecutionStatus.SUCCEEDED
 class WaitTask implements RetryableTask {
   long backoffPeriod = 15000
   long timeout = Integer.MAX_VALUE
+  long waitTimeMs
 
   TimeProvider timeProvider = new TimeProvider()
 
+  @Override
+  long getDynamicBackoffPeriod(Duration taskDuration) {
+    if (taskDuration <= Duration.ofMillis(waitTimeMs)) {
+      // wait until timeout is over to poll
+      return Duration.ofMillis(waitTimeMs).toMillis()
+    } else {
+      // start polling normally after timeout to account for delays like throttling
+      return backoffPeriod
+    }
+  }
+
+
   @Override
   TaskResult execute(Stage stage) {
     // wait time is specified in seconds
     long waitTime = stage.context.waitTime as long
-    def waitTimeMs = TimeUnit.MILLISECONDS.convert(waitTime, TimeUnit.SECONDS)
+    waitTimeMs = TimeUnit.MILLISECONDS.convert(waitTime, TimeUnit.SECONDS)
     def now = timeProvider.millis
 
     if (!stage.context.containsKey("waitTaskState") || !stage.context.waitTaskState instanceof Map) {

orca-echo/src/main/groovy/com/netflix/spinnaker/orca/echo/pipeline/ManualJudgmentStage.groovy

@@ -16,6 +16,7 @@
 
 package com.netflix.spinnaker.orca.echo.pipeline
 
+import java.time.Duration
 import java.util.concurrent.TimeUnit
 import com.google.common.annotations.VisibleForTesting
 import com.netflix.spinnaker.orca.*
@@ -65,11 +66,24 @@ class ManualJudgmentStage implements StageDefinitionBuilder, RestartableStage, A
     long backoffPeriod = 15000
     long timeout = TimeUnit.DAYS.toMillis(3)
 
+    @Override
+    long getDynamicBackoffPeriod(Duration taskDuration) {
+      if (taskDuration < Duration.ofMillis(timeout)) {
+        // wait until timeout is over to poll
+        return Duration.ofMillis(timeout).toMillis()
+      } else {
+        // start polling normally after timeout to account for delays like throttling
+        return backoffPeriod
+      }
+    }
+
     @Autowired(required = false)
     EchoService echoService
 
     @Override
     TaskResult execute(Stage stage) {
+      stage.getTopLevelTimeout().ifPresent({ timeout = it })
+
       StageData stageData = stage.mapTo(StageData)
       NotificationState notificationState
       ExecutionStatus executionStatus

orca-web/src/main/groovy/com/netflix/spinnaker/orca/controllers/TaskController.groovy

@@ -16,6 +16,9 @@
 
 package com.netflix.spinnaker.orca.controllers
 
+import com.netflix.spinnaker.orca.pipeline.OrchestrationLauncher
+import com.netflix.spinnaker.orca.pipeline.model.Task
+
 import java.time.Clock
 import com.netflix.spinnaker.orca.ExecutionStatus
 import com.netflix.spinnaker.orca.front50.Front50Service
@@ -262,6 +265,8 @@ class TaskController {
       stage.context["lastModifiedBy"] = AuthenticatedRequest.getSpinnakerUser().orElse("anonymous")
 
       executionRepository.storeStage(stage)
+
+      executionRunner.reschedule(pipeline)
     }
     pipeline
   }

orca-web/src/test/groovy/com/netflix/spinnaker/orca/controllers/TaskControllerSpec.groovy

@@ -16,6 +16,8 @@
 
 package com.netflix.spinnaker.orca.controllers
 
+import com.netflix.spinnaker.orca.pipeline.ExecutionRunner
+
 import java.time.Clock
 import java.time.Instant
 import com.fasterxml.jackson.databind.ObjectMapper
@@ -45,6 +47,7 @@ class TaskControllerSpec extends Specification {
   def executionRepository = Mock(ExecutionRepository)
   def front50Service = Mock(Front50Service)
   def startTracker = Mock(PipelineStartTracker)
+  def executionRunner = Mock(ExecutionRunner)
 
   def clock = Clock.fixed(Instant.now(), UTC)
   int daysOfExecutionHistory = 14
@@ -57,6 +60,7 @@ class TaskControllerSpec extends Specification {
       new TaskController(
         front50Service: front50Service,
         executionRepository: executionRepository,
+        executionRunner: executionRunner,
         daysOfExecutionHistory: daysOfExecutionHistory,
         numberOfOldPipelineExecutionsToInclude: numberOfOldPipelineExecutionsToInclude,
         startTracker: startTracker,
@@ -293,6 +297,7 @@ class TaskControllerSpec extends Specification {
         judgmentStatus: "stop", value: "1", lastModifiedBy: "anonymous"
       ]
     } as Stage)
+    1 * executionRunner.reschedule(pipeline)
     0 * _
 
     and: