<commit_before>package cromwell

import akka.testkit.{EventFilter, TestActorRef}
import cromwell.engine.backend._
import cromwell.engine.backend.local.LocalBackend
import cromwell.engine.workflow.WorkflowManagerActor
import cromwell.engine.workflow.WorkflowManagerActor._
import cromwell.engine.{PreemptedException, WorkflowId, WorkflowSucceeded}
import cromwell.util.SampleWdl
import cromwell.webservice.CromwellApiHandler._
import org.specs2.mock.Mockito
import wdl4s.values.{WdlValue, WdlArray}

import scala.concurrent.{ExecutionContext, Future}

class RetryableCallsSpec extends CromwellTestkitSpec with Mockito {
  val customizedLocalBackend = new LocalBackend(system) {
    override def execute(backendCall: BackendCall)(implicit ec: ExecutionContext): Future[ExecutionHandle] = {
      backendCall.key.scope.taskFqn match {
        case "do_scatter" if backendCall.key.index.contains(0) && backendCall.key.attempt == 1 =>
          RetryableExecutionHandle(new PreemptedException("Retryable failure")).future
        case _ => super.execute(backendCall)
      }
    }
  }

  "A workflow with a Successful retried Call" should {

    "succeed and return correct metadata" in {
      implicit val workflowManagerActor = TestActorRef(WorkflowManagerActor.props(customizedLocalBackend), self, "Test Workflow metadata with Retried Calls")

      val workflowId = waitForHandledMessagePattern(pattern = "transitioning from Running to Succeeded") {
        EventFilter.info(pattern = s"persisting status of do_scatter:0:2 to Starting", occurrences = 1).intercept {
          messageAndWait[WorkflowManagerSubmitSuccess](SubmitWorkflow(SampleWdl.PrepareScatterGatherWdl.asWorkflowSources())).id
        }
      }

      val status = messageAndWait[WorkflowManagerStatusSuccess](WorkflowStatus(workflowId)).state
      status shouldEqual WorkflowSucceeded

      val metadata = messageAndWait[WorkflowManagerWorkflowMetadataSuccess](WorkflowMetadata(workflowId)).response
      metadata should not be null

      metadata.status shouldBe WorkflowSucceeded.toString
      metadata.start shouldBe defined
      metadata.end shouldBe defined
      metadata.outputs shouldBe defined
      metadata.outputs.get should have size 3
      metadata.calls should have size 3
      metadata.calls.get("sc_test.do_scatter") shouldBe defined
      metadata.calls("sc_test.do_scatter") should have size 5

      metadata.calls("sc_test.do_scatter").head.executionStatus shouldBe "Preempted"
      metadata.calls("sc_test.do_scatter").head.shardIndex shouldBe 0
      metadata.calls("sc_test.do_scatter").head.attempt shouldBe 1
      metadata.calls("sc_test.do_scatter").head.stdout shouldBe defined
      metadata.calls("sc_test.do_scatter").head.stdout.get.value should not include "attempt"
      metadata.calls("sc_test.do_scatter").head.stderr shouldBe defined
      metadata.calls("sc_test.do_scatter").head.stderr.get.value should not include "attempt"
      metadata.calls("sc_test.do_scatter").head.outputs.get should have size 0
      metadata.calls("sc_test.do_scatter")(1).executionStatus shouldBe "Done"
      metadata.calls("sc_test.do_scatter")(1).shardIndex shouldBe 0
      metadata.calls("sc_test.do_scatter")(1).attempt shouldBe 2
      metadata.calls("sc_test.do_scatter")(1).stdout shouldBe defined
      metadata.calls("sc_test.do_scatter")(1).stdout.get.value should include ("attempt")
      metadata.calls("sc_test.do_scatter")(1).stderr shouldBe defined
      metadata.calls("sc_test.do_scatter")(1).stderr.get.value should include ("attempt")
      metadata.calls("sc_test.do_scatter")(1).outputs shouldBe defined
      metadata.calls("sc_test.do_scatter")(1).outputs.get should not be empty

      // Check that only successful shards (and not preempted ones) are collected
      val gatherInputs: Seq[WdlValue] = metadata.calls("sc_test.do_gather").head.inputs("input_files").asInstanceOf[WdlArray].value
      gatherInputs.size shouldBe 4

      val wfStdouterr = messageAndWait[WorkflowManagerWorkflowStdoutStderrSuccess](WorkflowStdoutStderr(workflowId))
      wfStdouterr should not be null
      wfStdouterr.logs should have size 3
      wfStdouterr.logs("sc_test.do_scatter") should have size 5
      wfStdouterr.logs("sc_test.do_scatter").head.stdout.value should not include "attempt"
      wfStdouterr.logs("sc_test.do_scatter").head.stderr.value should not include "attempt"
      wfStdouterr.logs("sc_test.do_scatter")(1).stdout.value should include ("attempt")
      wfStdouterr.logs("sc_test.do_scatter")(1).stderr.value should include ("attempt")

      val callStdouterr = messageAndWait[WorkflowManagerCallStdoutStderrSuccess](CallStdoutStderr(workflowId, "sc_test.do_scatter"))
      callStdouterr should not be null
      callStdouterr.logs should have size 5
      callStdouterr.logs.head.stdout.value should not include "attempt"
      callStdouterr.logs.head.stderr.value should not include "attempt"
      callStdouterr.logs(1).stdout.value should include ("attempt")
      callStdouterr.logs(1).stderr.value should include ("attempt")

    }

  }
}
<commit_msg>Disable call cache for Retryable Calls<commit_after>package cromwell

import akka.testkit.{EventFilter, TestActorRef}
import cromwell.engine.backend._
import cromwell.engine.backend.local.LocalBackend
import cromwell.engine.workflow.WorkflowManagerActor
import cromwell.engine.workflow.WorkflowManagerActor._
import cromwell.engine.{PreemptedException, WorkflowId, WorkflowSucceeded}
import cromwell.util.SampleWdl
import cromwell.webservice.CromwellApiHandler._
import org.specs2.mock.Mockito
import wdl4s.values.{WdlValue, WdlArray}

import scala.concurrent.{ExecutionContext, Future}

class RetryableCallsSpec extends CromwellTestkitSpec with Mockito {
  val customizedLocalBackend = new LocalBackend(system) {
    override def execute(backendCall: BackendCall)(implicit ec: ExecutionContext): Future[ExecutionHandle] = {
      backendCall.key.scope.taskFqn match {
        case "do_scatter" if backendCall.key.index.contains(0) && backendCall.key.attempt == 1 =>
          RetryableExecutionHandle(new PreemptedException("Retryable failure")).future
        case _ => super.execute(backendCall)
      }
    }
  }

  "A workflow with a Successful retried Call" should {

    "succeed and return correct metadata" in {
      implicit val workflowManagerActor = TestActorRef(WorkflowManagerActor.props(customizedLocalBackend), self, "Test Workflow metadata with Retried Calls")
      val noCallCaching = """{ "read-from-cache": false }"""
      val workflowId = waitForHandledMessagePattern(pattern = "transitioning from Running to Succeeded") {
        EventFilter.info(pattern = s"persisting status of do_scatter:0:2 to Starting", occurrences = 1).intercept {
          messageAndWait[WorkflowManagerSubmitSuccess](SubmitWorkflow(SampleWdl.PrepareScatterGatherWdl.asWorkflowSources(workflowOptions = noCallCaching))).id
        }
      }

      val status = messageAndWait[WorkflowManagerStatusSuccess](WorkflowStatus(workflowId)).state
      status shouldEqual WorkflowSucceeded

      val metadata = messageAndWait[WorkflowManagerWorkflowMetadataSuccess](WorkflowMetadata(workflowId)).response
      metadata should not be null

      metadata.status shouldBe WorkflowSucceeded.toString
      metadata.start shouldBe defined
      metadata.end shouldBe defined
      metadata.outputs shouldBe defined
      metadata.outputs.get should have size 3
      metadata.calls should have size 3
      metadata.calls.get("sc_test.do_scatter") shouldBe defined
      metadata.calls("sc_test.do_scatter") should have size 5

      metadata.calls("sc_test.do_scatter").head.executionStatus shouldBe "Preempted"
      metadata.calls("sc_test.do_scatter").head.shardIndex shouldBe 0
      metadata.calls("sc_test.do_scatter").head.attempt shouldBe 1
      metadata.calls("sc_test.do_scatter").head.stdout shouldBe defined
      metadata.calls("sc_test.do_scatter").head.stdout.get.value should not include "attempt"
      metadata.calls("sc_test.do_scatter").head.stderr shouldBe defined
      metadata.calls("sc_test.do_scatter").head.stderr.get.value should not include "attempt"
      metadata.calls("sc_test.do_scatter").head.outputs.get should have size 0
      metadata.calls("sc_test.do_scatter")(1).executionStatus shouldBe "Done"
      metadata.calls("sc_test.do_scatter")(1).shardIndex shouldBe 0
      metadata.calls("sc_test.do_scatter")(1).attempt shouldBe 2
      metadata.calls("sc_test.do_scatter")(1).stdout shouldBe defined
      metadata.calls("sc_test.do_scatter")(1).stdout.get.value should include ("attempt")
      metadata.calls("sc_test.do_scatter")(1).stderr shouldBe defined
      metadata.calls("sc_test.do_scatter")(1).stderr.get.value should include ("attempt")
      metadata.calls("sc_test.do_scatter")(1).outputs shouldBe defined
      metadata.calls("sc_test.do_scatter")(1).outputs.get should not be empty

      // Check that only successful shards (and not preempted ones) are collected
      val gatherInputs: Seq[WdlValue] = metadata.calls("sc_test.do_gather").head.inputs("input_files").asInstanceOf[WdlArray].value
      gatherInputs.size shouldBe 4

      val wfStdouterr = messageAndWait[WorkflowManagerWorkflowStdoutStderrSuccess](WorkflowStdoutStderr(workflowId))
      wfStdouterr should not be null
      wfStdouterr.logs should have size 3
      wfStdouterr.logs("sc_test.do_scatter") should have size 5
      wfStdouterr.logs("sc_test.do_scatter").head.stdout.value should not include "attempt"
      wfStdouterr.logs("sc_test.do_scatter").head.stderr.value should not include "attempt"
      wfStdouterr.logs("sc_test.do_scatter")(1).stdout.value should include ("attempt")
      wfStdouterr.logs("sc_test.do_scatter")(1).stderr.value should include ("attempt")

      val callStdouterr = messageAndWait[WorkflowManagerCallStdoutStderrSuccess](CallStdoutStderr(workflowId, "sc_test.do_scatter"))
      callStdouterr should not be null
      callStdouterr.logs should have size 5
      callStdouterr.logs.head.stdout.value should not include "attempt"
      callStdouterr.logs.head.stderr.value should not include "attempt"
      callStdouterr.logs(1).stdout.value should include ("attempt")
      callStdouterr.logs(1).stderr.value should include ("attempt")

    }

  }
}

{'fraction_non_alphanumeric': 0.09571373157486424, 'fraction_numerical': 0.0062063615205585725, 'mean_word_length': 4.623227917121047, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '818941', 'n_tokens_mistral': 3123, 'n_tokens_neox': 3058, 'n_words': 596}