diff --git a/.github/workflows/chart_update_on_merge.yml b/.github/workflows/chart_update_on_merge.yml
index d92ee8ef102..2788681fd08 100644
--- a/.github/workflows/chart_update_on_merge.yml
+++ b/.github/workflows/chart_update_on_merge.yml
@@ -11,6 +11,7 @@ jobs:
     if: github.event.pull_request.merged == true
     runs-on: ubuntu-latest
     steps:
+    - uses: sbt/setup-sbt@v1
     - name: Clone Cromwell
       uses: actions/checkout@v4
       with:
diff --git a/.github/workflows/cromwell_unit_tests.yml b/.github/workflows/cromwell_unit_tests.yml
index 88951871d8f..776685f7e98 100644
--- a/.github/workflows/cromwell_unit_tests.yml
+++ b/.github/workflows/cromwell_unit_tests.yml
@@ -22,6 +22,7 @@ jobs:
 
     steps:
     - uses: actions/checkout@v3 # checkout the cromwell repo
+    - uses: sbt/setup-sbt@v1
     - uses: ./.github/set_up_cromwell_action #Exectute this reusable github action. It will set up java/sbt/git-secrets/cromwell.
       with:
         cromwell_repo_token: ${{ secrets.BROADBOT_GITHUB_TOKEN }}
diff --git a/.github/workflows/docker_build_test.yml b/.github/workflows/docker_build_test.yml
index d3e5369fca5..2ea9d9a4970 100644
--- a/.github/workflows/docker_build_test.yml
+++ b/.github/workflows/docker_build_test.yml
@@ -25,6 +25,7 @@ jobs:
           repository: broadinstitute/cromwell
           token: ${{ secrets.BROADBOT_GITHUB_TOKEN }}
           path: cromwell
+      - uses: sbt/setup-sbt@v1
       - uses: actions/setup-java@v4
         with:
           distribution: 'temurin'
diff --git a/.github/workflows/integration_tests.yml b/.github/workflows/integration_tests.yml
index bf13316158a..d898fa222ef 100644
--- a/.github/workflows/integration_tests.yml
+++ b/.github/workflows/integration_tests.yml
@@ -93,9 +93,9 @@ jobs:
             friendly_name: Centaur Engine Upgrade Local with MySQL 5.7
           - build_type: referenceDiskManifestBuilderApp
             friendly_name: Reference Disk Manifest Builder App
-          - build_type: centaurSlurm
-            build_mysql: 5.7
-            friendly_name: "Centaur Slurm with MySQL 5.7"
+#          - build_type: centaurSlurm
+#            build_mysql: 5.7
+#            friendly_name: "Centaur Slurm with MySQL 5.7"
           - build_type: centaurBlob
             build_mysql: 5.7
             friendly_name: Centaur Blob
@@ -114,6 +114,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 120
     steps:
+    - uses: sbt/setup-sbt@v1
     - uses: actions/checkout@v3 # checkout the cromwell repo
       with:
         ref: ${{ inputs.target-branch }}
diff --git a/.github/workflows/scalafmt-check.yml b/.github/workflows/scalafmt-check.yml
index 3730d2ffc8f..e547fcc85a4 100644
--- a/.github/workflows/scalafmt-check.yml
+++ b/.github/workflows/scalafmt-check.yml
@@ -19,6 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 20
     steps:
+      - uses: sbt/setup-sbt@v1
       - uses: actions/checkout@v3
         with:
           ref: ${{ inputs.target-branch }}
diff --git a/.github/workflows/trivy.yml b/.github/workflows/trivy.yml
index 1704b3b826c..0e190503d22 100644
--- a/.github/workflows/trivy.yml
+++ b/.github/workflows/trivy.yml
@@ -17,6 +17,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v2
+      - uses: sbt/setup-sbt@v1
 
       # fetch SBT package
       - uses: actions/setup-java@v4
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 36abc519ed7..3f5fa58aaa8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -55,9 +55,17 @@ The `IX_WORKFLOW_STORE_ENTRY_WS` index is removed from `WORKFLOW_STORE_ENTRY`.
 
 The index had low cardinality and workflow pickup is faster without it. Migration time depends on workflow store size, but should be very fast for most installations. Terminal workflows are removed from the workflow store, so only running workflows contribute to the cost.
 
-### Bug fixes and small changes
+#### Index additions
 
- * Changed default boot disk size from 10GB to 20GB in PipelinesAPI and Google Batch backends
+The `IX_METADATA_ENTRY_WEU_MK` index is added to `METADATA_ENTRY`. In pre-release testing, the migration proceeded at about 3 million rows per minute. Please plan downtime accordingly.
+
+#### Unique constraint addition
+
+The `UC_GROUP_METRICS_ENTRY_GI` constraint has been added to column `GROUP_ID` in `GROUP_METRICS_ENTRY` table. This additionally enforces at database level that the `GROUP_ID` column always contains unique values.
+
+### Reduce errors from boot disk filling up on Google Lifesciences API
+
+ * If Cromwell can't determine the size of the user command Docker image, it will increase Lifesciences API boot disk size by 30GB rather than 0. This should reduce incidence of tasks failing due to boot disk filling up.
 
 #### Improved `size()` function performance on arrays
 
diff --git a/centaur/src/main/resources/standardTestCases/docker_size_dockerhub.test b/centaur/src/main/resources/standardTestCases/docker_size_dockerhub.test
index f7d59e4b390..dc3d297f6fc 100644
--- a/centaur/src/main/resources/standardTestCases/docker_size_dockerhub.test
+++ b/centaur/src/main/resources/standardTestCases/docker_size_dockerhub.test
@@ -11,8 +11,8 @@ files {
 
 metadata {
   status: Succeeded
-  "outputs.docker_size_dockerhub.large_dockerhub_image_with_hash.bootDiskSize": 27
-  "outputs.docker_size_dockerhub.large_dockerhub_image_with_tag.bootDiskSize": 27
+  "outputs.docker_size_dockerhub.large_dockerhub_image_with_hash.bootDiskSize": 17
+  "outputs.docker_size_dockerhub.large_dockerhub_image_with_tag.bootDiskSize": 17
 }
 
 workflowType: WDL
diff --git a/centaur/src/main/resources/standardTestCases/docker_size_gcr.test b/centaur/src/main/resources/standardTestCases/docker_size_gcr.test
index 2e8c2e1b2d3..7399089f307 100644
--- a/centaur/src/main/resources/standardTestCases/docker_size_gcr.test
+++ b/centaur/src/main/resources/standardTestCases/docker_size_gcr.test
@@ -11,8 +11,8 @@ files {
 
 metadata {
   status: Succeeded
-  "outputs.docker_size_gcr.large_gcr_image_with_hash.bootDiskSize": 27
-  "outputs.docker_size_gcr.large_gcr_image_with_tag.bootDiskSize": 27
+  "outputs.docker_size_gcr.large_gcr_image_with_hash.bootDiskSize": 17
+  "outputs.docker_size_gcr.large_gcr_image_with_tag.bootDiskSize": 17
 }
 
 workflowType: WDL
diff --git a/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureConfiguration.scala b/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureConfiguration.scala
new file mode 100644
index 00000000000..6a3d4acaa62
--- /dev/null
+++ b/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureConfiguration.scala
@@ -0,0 +1,28 @@
+package cromwell.cloudsupport.azure
+
+import com.azure.core.management.AzureEnvironment
+import com.typesafe.config.ConfigFactory
+import net.ceedubs.ficus.Ficus._
+
+object AzureConfiguration {
+  private val conf = ConfigFactory.load().getConfig("azure")
+  val azureEnvironment =
+    AzureEnvironmentConverter.fromString(
+      conf.as[Option[String]]("azure-environment").getOrElse(AzureEnvironmentConverter.Azure)
+    )
+  val azureTokenScopeManagement = conf.as[String]("token-scope-management")
+}
+
+object AzureEnvironmentConverter {
+  val Azure: String = "AzureCloud"
+  val AzureGov: String = "AzureUSGovernmentCloud"
+  val AzureChina: String = "AzureChinaCloud"
+
+  def fromString(s: String): AzureEnvironment = s match {
+    case AzureGov => AzureEnvironment.AZURE_US_GOVERNMENT
+    case AzureChina => AzureEnvironment.AZURE_CHINA
+    // a bit redundant, but I want to have a explicit case for Azure for clarity, even though it's the default
+    case Azure => AzureEnvironment.AZURE
+    case _ => AzureEnvironment.AZURE
+  }
+}
diff --git a/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureCredentials.scala b/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureCredentials.scala
index d3d66e1bafc..034e464aa54 100644
--- a/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureCredentials.scala
+++ b/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureCredentials.scala
@@ -2,7 +2,6 @@ package cromwell.cloudsupport.azure
 
 import cats.implicits.catsSyntaxValidatedId
 import com.azure.core.credential.TokenRequestContext
-import com.azure.core.management.AzureEnvironment
 import com.azure.core.management.profile.AzureProfile
 import com.azure.identity.DefaultAzureCredentialBuilder
 import common.validation.ErrorOr.ErrorOr
@@ -20,8 +19,8 @@ case object AzureCredentials {
 
   final val tokenAcquisitionTimeout = 5.seconds
 
-  val azureProfile = new AzureProfile(AzureEnvironment.AZURE)
-  val tokenScope = "https://management.azure.com/.default"
+  val azureProfile = new AzureProfile(AzureConfiguration.azureEnvironment)
+  val tokenScope = AzureConfiguration.azureTokenScopeManagement
 
   private def tokenRequestContext: TokenRequestContext = {
     val trc = new TokenRequestContext()
diff --git a/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureUtils.scala b/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureUtils.scala
index dd379ed3564..1febd5d77c0 100644
--- a/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureUtils.scala
+++ b/cloudSupport/src/main/scala/cromwell/cloudsupport/azure/AzureUtils.scala
@@ -1,6 +1,5 @@
 package cromwell.cloudsupport.azure
 
-import com.azure.core.management.AzureEnvironment
 import com.azure.core.management.profile.AzureProfile
 import com.azure.identity.DefaultAzureCredentialBuilder
 import com.azure.resourcemanager.AzureResourceManager
@@ -33,7 +32,7 @@ object AzureUtils {
       .map(Success(_))
       .getOrElse(Failure(new Exception("Could not parse storage account")))
 
-    val azureProfile = new AzureProfile(AzureEnvironment.AZURE)
+    val azureProfile = new AzureProfile(AzureConfiguration.azureEnvironment)
 
     def azureCredentialBuilder = new DefaultAzureCredentialBuilder()
       .authorityHost(azureProfile.getEnvironment.getActiveDirectoryEndpoint)
diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf
index 9da59f16b7b..057b20a9642 100644
--- a/core/src/main/resources/reference.conf
+++ b/core/src/main/resources/reference.conf
@@ -330,6 +330,11 @@ call-caching {
   max-failed-copy-attempts = 1000000
 }
 
+azure {
+    azure-environment = "AzureCloud"
+    token-scope-management = "https://management.azure.com/.default"
+}
+
 google {
 
   application-name = "cromwell"
diff --git a/database/migration/src/main/resources/changelog.xml b/database/migration/src/main/resources/changelog.xml
index 24c1943eec7..81036465176 100644
--- a/database/migration/src/main/resources/changelog.xml
+++ b/database/migration/src/main/resources/changelog.xml
@@ -92,6 +92,7 @@
     <include file="changesets/enlarge_sub_workflow_store_entry_id.xml" relativeToChangelogFile="true" />
     <include file="changesets/workflow_store_drop_state_index.xml" relativeToChangelogFile="true" />
     <include file="changesets/add_group_metrics_table.xml" relativeToChangelogFile="true" />
+    <include file="changesets/add_group_metrics_unique_constraint.xml" relativeToChangelogFile="true" />
     <!-- WARNING!
       This changeset should always be last.
       It is always run (and should always run last) to set table ownership correctly.
diff --git a/database/migration/src/main/resources/changesets/add_group_metrics_unique_constraint.xml b/database/migration/src/main/resources/changesets/add_group_metrics_unique_constraint.xml
new file mode 100644
index 00000000000..14574ef98b0
--- /dev/null
+++ b/database/migration/src/main/resources/changesets/add_group_metrics_unique_constraint.xml
@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<databaseChangeLog objectQuotingStrategy="QUOTE_ALL_OBJECTS"
+                   xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.3.xsd">
+
+<changeSet id="add_unique_constraint_group_metrics" author="sshah" dbms="mysql,hsqldb,mariadb,postgresql">
+    <dropIndex
+            tableName="GROUP_METRICS_ENTRY"
+            indexName="IX_GROUP_METRICS_ENTRY_GI" />
+
+    <addUniqueConstraint
+            tableName="GROUP_METRICS_ENTRY"
+            columnNames="GROUP_ID"
+            constraintName="UC_GROUP_METRICS_ENTRY_GI"/>
+    </changeSet>
+</databaseChangeLog>
diff --git a/database/migration/src/main/resources/metadata_changesets/metadata_index_add_workflow_key.xml b/database/migration/src/main/resources/metadata_changesets/metadata_index_add_workflow_key.xml
new file mode 100644
index 00000000000..98171be9d72
--- /dev/null
+++ b/database/migration/src/main/resources/metadata_changesets/metadata_index_add_workflow_key.xml
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<databaseChangeLog objectQuotingStrategy="QUOTE_ALL_OBJECTS"
+                   xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.3.xsd">
+
+    <changeSet id="metadata_index_add_workflow_key" author="anichols" dbms="hsqldb,mariadb,mysql,postgresql">
+        <!--
+            This index creates at about 3M rows per minute on MySQL.
+            That would be an impossible multi-day downtime in Terra, so we manually pre-create the index asynchronously.
+            This changeset detects environments where this has been done and immediately marks itself as applied.
+        -->
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <indexExists indexName="IX_METADATA_ENTRY_WEU_MK"/>
+            </not>
+        </preConditions>
+        <createIndex indexName="IX_METADATA_ENTRY_WEU_MK" tableName="METADATA_ENTRY">
+            <column name="WORKFLOW_EXECUTION_UUID"/>
+            <column name="METADATA_KEY"/>
+        </createIndex>
+    </changeSet>
+</databaseChangeLog>
diff --git a/database/migration/src/main/resources/sql_metadata_changelog.xml b/database/migration/src/main/resources/sql_metadata_changelog.xml
index 0989ec2199a..11437a2ce3a 100644
--- a/database/migration/src/main/resources/sql_metadata_changelog.xml
+++ b/database/migration/src/main/resources/sql_metadata_changelog.xml
@@ -19,6 +19,7 @@
     <include file="metadata_changesets/remove_non_summarizable_metadata_from_queue.xml" relativeToChangelogFile="true" />
     <include file="metadata_changesets/update_metadata_archive_index.xml" relativeToChangelogFile="true" />
     <include file="metadata_changesets/reset_archive_statuses_to_null.xml" relativeToChangelogFile="true" />
+    <include file="metadata_changesets/metadata_index_add_workflow_key.xml" relativeToChangelogFile="true" />
     <!-- WARNING!
       This changeset should always be last.
       It it always run (and should always run last) to set table ownership correctly.
diff --git a/database/sql/src/main/scala/cromwell/database/slick/GroupMetricsSlickDatabase.scala b/database/sql/src/main/scala/cromwell/database/slick/GroupMetricsSlickDatabase.scala
index ee73873b9f7..af7a7d694c0 100644
--- a/database/sql/src/main/scala/cromwell/database/slick/GroupMetricsSlickDatabase.scala
+++ b/database/sql/src/main/scala/cromwell/database/slick/GroupMetricsSlickDatabase.scala
@@ -2,8 +2,10 @@ package cromwell.database.slick
 
 import cromwell.database.sql.GroupMetricsSqlDatabase
 import cromwell.database.sql.tables.GroupMetricsEntry
+import org.postgresql.util.PSQLException
+import slick.jdbc.TransactionIsolation
 
-import java.sql.Timestamp
+import java.sql.{SQLIntegrityConstraintViolationException, Timestamp}
 import scala.concurrent.{ExecutionContext, Future}
 
 trait GroupMetricsSlickDatabase extends GroupMetricsSqlDatabase {
@@ -14,16 +16,47 @@ trait GroupMetricsSlickDatabase extends GroupMetricsSqlDatabase {
   override def recordGroupMetricsEntry(
     groupMetricsEntry: GroupMetricsEntry
   )(implicit ec: ExecutionContext): Future[Unit] = {
-    val action = for {
-      updateCount <- dataAccess
-        .quotaExhaustionForGroupId(groupMetricsEntry.groupId)
-        .update(groupMetricsEntry.quotaExhaustionDetected)
-      _ <- updateCount match {
-        case 0 => dataAccess.groupMetricsEntryIdsAutoInc += groupMetricsEntry
-        case _ => assertUpdateCount("recordGroupMetricsEntry", updateCount, 1)
-      }
+    def updateGroupMetricsEntry(): Future[Unit] = {
+      val updateAction = for {
+        _ <- dataAccess
+          .quotaExhaustionForGroupId(groupMetricsEntry.groupId)
+          .update(groupMetricsEntry.quotaExhaustionDetected)
+      } yield ()
+
+      // The transaction level is set to 'ReadCommitted' to avoid Postgres returning error `could not serialize
+      // access due to concurrent update` which happens in 'RepeatableRead' or higher isolation levels.
+      // See https://stackoverflow.com/questions/50797097/postgres-could-not-serialize-access-due-to-concurrent-update
+      runTransaction(updateAction, TransactionIsolation.ReadCommitted)
+    }
+
+    /*
+      The approach here is to try and insert the record into the table and if that fails because a record with that
+      group_id already exists, then it will update that record with new quota_exhaustion_detected timestamp.
+      The insert and update happen in 2 different transactions. Using 2 separate transactions avoids deadlocks that
+      occur during upserts or when using Slick's insertOrUpdate in a single transaction. Deadlocks occur in scenarios
+      with concurrent threads trying to update the table, even under the stricter Serializable transaction isolation
+      level. This is caused by a gap lock on the IX_GROUP_METRICS_ENTRY_GI index, which locks gaps between index
+      records on a page, leading to deadlocks for transactions involving the same or nearby group_id values.
+      See https://broadworkbench.atlassian.net/browse/AN-286 and https://broadworkbench.atlassian.net/browse/WX-1847.
+
+      This approach should have minimal performance impact since the table has only 3 columns and low cardinality.
+
+      Note: a unique constraint on group_id also exists in database.
+     */
+    val insertAction = for {
+      _ <- dataAccess.groupMetricsEntryIdsAutoInc += groupMetricsEntry
     } yield ()
-    runTransaction(action)
+
+    runTransaction(insertAction)
+      .recoverWith {
+        case ex
+            if ex.isInstanceOf[SQLIntegrityConstraintViolationException] || (ex
+              .isInstanceOf[PSQLException] && ex.getMessage.contains(
+              "duplicate key value violates unique constraint \"UC_GROUP_METRICS_ENTRY_GI\""
+            )) =>
+          updateGroupMetricsEntry()
+        case e => Future.failed(e)
+      }
   }
 
   override def countGroupMetricsEntries(groupId: String)(implicit ec: ExecutionContext): Future[Int] = {
diff --git a/database/sql/src/main/scala/cromwell/database/slick/tables/GroupMetricsEntryComponent.scala b/database/sql/src/main/scala/cromwell/database/slick/tables/GroupMetricsEntryComponent.scala
index 4b8844746e3..ed5b708d3ad 100644
--- a/database/sql/src/main/scala/cromwell/database/slick/tables/GroupMetricsEntryComponent.scala
+++ b/database/sql/src/main/scala/cromwell/database/slick/tables/GroupMetricsEntryComponent.scala
@@ -31,7 +31,7 @@ trait GroupMetricsEntryComponent {
                                                        groupMetricsEntryId.?
     ) <> ((GroupMetricsEntry.apply _).tupled, GroupMetricsEntry.unapply)
 
-    def ixGroupMetricsEntryGi = index("IX_GROUP_METRICS_ENTRY_GI", groupId, unique = false)
+    def ucGroupMetricsEntryGi = index("UC_GROUP_METRICS_ENTRY_GI", groupId, unique = true)
   }
 
   protected val groupMetricsEntries = TableQuery[GroupMetricsEntries]
diff --git a/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala b/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala
index 0ca87f4e2d3..de6d2c838df 100644
--- a/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala
+++ b/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala
@@ -57,6 +57,23 @@ trait MetadataEntryComponent {
 
     // TODO: rename index via liquibase
     def ixMetadataEntryWeu = index("METADATA_WORKFLOW_IDX", workflowExecutionUuid, unique = false)
+
+    /**
+      * Index designed to accelerate common key-specific queries across an entire workflow, such as:
+      * - Get workflow-level `outputs%` keys (no tasks, requireEmptyJobKey = true)
+      * - Get all `vmStartTime%`, `vmEndTime%`, `vmCostPerHour%` keys in the workflow (include tasks, requireEmptyJobKey = false)
+      *
+      * It is NOT good, as in may make actively slower, queries that reference a specific job. If we do more
+      * with getting metadata for individual jobs, recommend creating this index with all 5 columns:
+      * - WORKFLOW_EXECUTION_UUID, CALL_FQN, JOB_SCATTER_INDEX, JOB_RETRY_ATTEMPT, METADATA_KEY
+      *
+      * Do NOT recommend this alternate order, as wildcards in the middle are inefficient and this can be
+      * slower than no indexes. Tested with 20M row `69e8259c` workflow in October 2024.
+      * - WORKFLOW_EXECUTION_UUID, METADATA_KEY, CALL_FQN, JOB_SCATTER_INDEX, JOB_RETRY_ATTEMPT
+      *
+      * @return A reference to the index
+      */
+    def ixMetadataEntryWeuMk = index("IX_METADATA_ENTRY_WEU_MK", (workflowExecutionUuid, metadataKey), unique = false)
   }
 
   val metadataEntries = TableQuery[MetadataEntries]
diff --git a/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala b/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala
index 02ffaefa3b7..87357af314c 100644
--- a/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala
+++ b/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala
@@ -347,7 +347,8 @@ class MetadataBuilderActorSpec
     val subWorkflow1aId = WorkflowId(UUID.fromString("1a1a1a1a-f76d-4af3-b371-5ba580916729"))
     val subWorkflow1bId = WorkflowId(UUID.fromString("1b1b1b1b-f76d-4af3-b371-5ba580916729"))
 
-    val workflowState = WorkflowSucceeded
+    val workflowSucceededState = WorkflowSucceeded
+    val workflowRunningState = WorkflowRunning
 
     val mainEvents = List(
       MetadataEvent(MetadataKey(mainWorkflowId, Option(MetadataJobKey("wfMain", None, 1)), "subWorkflowId"),
@@ -391,19 +392,25 @@ class MetadataBuilderActorSpec
     class TestReadDatabaseMetadataWorkerActorForCost extends ReadDatabaseMetadataWorkerActor(defaultTimeout, 1000000) {
       override def receive: Receive = {
         case GetCost(wfId) if wfId == mainWorkflowId =>
-          sender() ! CostResponse(mainWorkflowId, workflowState, MetadataLookupResponse(mainQuery, mainEvents))
+          sender() ! CostResponse(mainWorkflowId, workflowRunningState, MetadataLookupResponse(mainQuery, mainEvents))
           ()
         case GetCost(wfId) if wfId == subWorkflow1Id =>
-          sender() ! CostResponse(subWorkflow1Id, workflowState, MetadataLookupResponse(sub1Query, sub1Events))
+          sender() ! CostResponse(subWorkflow1Id, workflowSucceededState, MetadataLookupResponse(sub1Query, sub1Events))
           ()
         case GetCost(wfId) if wfId == subWorkflow2Id =>
-          sender() ! CostResponse(subWorkflow2Id, workflowState, MetadataLookupResponse(sub2Query, sub2Events))
+          sender() ! CostResponse(subWorkflow2Id, workflowSucceededState, MetadataLookupResponse(sub2Query, sub2Events))
           ()
         case GetCost(wfId) if wfId == subWorkflow1aId =>
-          sender() ! CostResponse(subWorkflow1aId, workflowState, MetadataLookupResponse(sub1aQuery, sub1aEvents))
+          sender() ! CostResponse(subWorkflow1aId,
+                                  workflowSucceededState,
+                                  MetadataLookupResponse(sub1aQuery, sub1aEvents)
+          )
           ()
         case GetCost(wfId) if wfId == subWorkflow1bId =>
-          sender() ! CostResponse(subWorkflow1bId, workflowState, MetadataLookupResponse(sub1bQuery, sub1bEvents))
+          sender() ! CostResponse(subWorkflow1bId,
+                                  workflowSucceededState,
+                                  MetadataLookupResponse(sub1bQuery, sub1bEvents)
+          )
           ()
         case _ => ()
       }
@@ -414,7 +421,7 @@ class MetadataBuilderActorSpec
          |"cost": 7,
          |"currency": "USD",
          |"id": "${mainWorkflowId}",
-         |"status": "${workflowState.toString}",
+         |"status": "${workflowRunningState.toString}",
          |"errors": ["Couldn't find valid vmCostPerHour for call1aA.-1.1"]
          |}""".stripMargin
 
diff --git a/filesystems/blob/src/main/scala/cromwell/filesystems/blob/BlobPathBuilder.scala b/filesystems/blob/src/main/scala/cromwell/filesystems/blob/BlobPathBuilder.scala
index 3c984d1f2c8..c4cf9a1d53c 100644
--- a/filesystems/blob/src/main/scala/cromwell/filesystems/blob/BlobPathBuilder.scala
+++ b/filesystems/blob/src/main/scala/cromwell/filesystems/blob/BlobPathBuilder.scala
@@ -3,6 +3,7 @@ package cromwell.filesystems.blob
 import akka.http.scaladsl.model.Uri
 import com.azure.storage.blob.nio.AzureBlobFileAttributes
 import com.google.common.net.UrlEscapers
+import cromwell.cloudsupport.azure.AzureConfiguration
 import cromwell.core.path.{NioPath, Path, PathBuilder}
 import cromwell.filesystems.blob.BlobPathBuilder._
 
@@ -13,13 +14,13 @@ import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
 
 object BlobPathBuilder {
-  private val blobHostnameSuffix = ".blob.core.windows.net"
+  private val blobHostnameSuffix = s".blob${AzureConfiguration.azureEnvironment.getStorageEndpointSuffix}"
   sealed trait BlobPathValidation
   case class ValidBlobPath(path: String, container: BlobContainerName, endpoint: EndpointURL) extends BlobPathValidation
   case class UnparsableBlobPath(errorMessage: Throwable) extends BlobPathValidation
 
   def invalidBlobHostMessage(endpoint: EndpointURL) =
-    s"Malformed Blob URL for this builder: The endpoint $endpoint doesn't contain the expected host string '{SA}.blob.core.windows.net/'"
+    s"Malformed Blob URL for this builder: The endpoint $endpoint doesn't contain the expected host string '{SA}.${blobHostnameSuffix}/'"
   def invalidBlobContainerMessage(endpoint: EndpointURL) =
     s"Malformed Blob URL for this builder: Could not parse container"
   val externalToken =
@@ -103,7 +104,8 @@ object BlobPath {
   // 1) If the path starts with http:/ (single slash!) transform it to the containerName:<path inside container>
   //    format the library expects
   // 2) If the path looks like <container>:<path>, strip off the <container>: to leave the absolute path inside the container.
-  private val brokenPathRegex = "https:/([a-z0-9]+).blob.core.windows.net/([-a-zA-Z0-9]+)/(.*)".r
+  private val brokenPathRegex =
+    s"https:/([a-z0-9]+).blob${AzureConfiguration.azureEnvironment.getStorageEndpointSuffix}/([-a-zA-Z0-9]+)/(.*)".r
 
   // Blob files larger than 5 GB upload in parallel parts [0][1] and do not get a native `CONTENT-MD5` property.
   // Instead, some uploaders such as TES [2] calculate the md5 themselves and store it under this key in metadata.
diff --git a/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala b/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala
index c337676e5b7..791323cf772 100644
--- a/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala
+++ b/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala
@@ -35,17 +35,43 @@ object MetadataBuilderActor {
   case object IdleData extends MetadataBuilderActorData
   final case class HasWorkData(target: ActorRef, originalRequest: BuildMetadataJsonAction)
       extends MetadataBuilderActorData
+
+  // Classes extending this trait are used to track state when the actor has launched child
+  // actors to collect metadata for subworkflows. This class aggregates data as it comes in,
+  // and builds the complete output when all subworkflow data is present. There's one child
+  // class for plain metadata queries and one for cost queries.
+  sealed trait EventsCollectorData extends MetadataBuilderActorData {
+    val target: ActorRef
+    val originalRequest: BuildMetadataJsonAction
+    val originalQuery: MetadataQuery
+    val originalEvents: Seq[MetadataEvent]
+    val subWorkflowsMetadata: Map[String, JsValue]
+    val waitFor: Int
+
+    def isComplete = subWorkflowsMetadata.size == waitFor
+  }
+
   final case class HasReceivedEventsData(target: ActorRef,
                                          originalRequest: BuildMetadataJsonAction,
                                          originalQuery: MetadataQuery,
                                          originalEvents: Seq[MetadataEvent],
                                          subWorkflowsMetadata: Map[String, JsValue],
                                          waitFor: Int
-  ) extends MetadataBuilderActorData {
+  ) extends EventsCollectorData {
     def withSubWorkflow(id: String, metadata: JsValue) =
       this.copy(subWorkflowsMetadata = subWorkflowsMetadata + ((id, metadata)))
+  }
 
-    def isComplete = subWorkflowsMetadata.size == waitFor
+  final case class HasReceivedCostEventsData(target: ActorRef,
+                                             originalRequest: BuildMetadataJsonAction,
+                                             originalQuery: MetadataQuery,
+                                             originalEvents: Seq[MetadataEvent],
+                                             originalStatus: WorkflowState,
+                                             subWorkflowsMetadata: Map[String, JsValue],
+                                             waitFor: Int
+  ) extends EventsCollectorData {
+    def withSubWorkflow(id: String, metadata: JsValue) =
+      this.copy(subWorkflowsMetadata = subWorkflowsMetadata + ((id, metadata)))
   }
 
   def props(readMetadataWorkerMaker: () => Props,
@@ -432,7 +458,7 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props,
   }
 
   when(WaitingForSubWorkflowCost) {
-    case Event(mbr: MetadataJsonResponse, data: HasReceivedEventsData) =>
+    case Event(mbr: MetadataJsonResponse, data: HasReceivedCostEventsData) =>
       processSubWorkflowCost(mbr, data)
     case Event(failure: MetadataServiceFailure, data: HasReceivedEventsData) =>
       data.target ! FailedMetadataJsonResponse(data.originalRequest, failure.reason)
@@ -482,7 +508,7 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props,
         failAndDie(new Exception(message), data.target, data.originalRequest)
     }
 
-  def processSubWorkflowCost(metadataResponse: MetadataJsonResponse, data: HasReceivedEventsData) =
+  def processSubWorkflowCost(metadataResponse: MetadataJsonResponse, data: HasReceivedCostEventsData) =
     metadataResponse match {
       case SuccessfulMetadataJsonResponse(GetCost(workflowId), js) =>
         val subId: WorkflowId = workflowId
@@ -491,7 +517,7 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props,
         if (newData.isComplete) {
           buildCostAndStop(
             data.originalQuery.workflowId,
-            extractFromJsAs[JsString](js, "status").map(_.value).getOrElse(""), // should never be empty
+            data.originalStatus,
             data.originalEvents,
             newData.subWorkflowsMetadata,
             data.target,
@@ -579,7 +605,7 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props,
 
     if (subWorkflowIds.isEmpty)
       // If no subworkflows found, just build cost data
-      buildCostAndStop(id, status.toString, metadataResponse.eventList, Map.empty, target, originalRequest)
+      buildCostAndStop(id, status, metadataResponse.eventList, Map.empty, target, originalRequest)
     else {
       // Otherwise spin up a metadata builder actor for each sub workflow
       subWorkflowIds foreach { subId =>
@@ -591,18 +617,19 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props,
         )
         subMetadataBuilder ! GetCost(WorkflowId.fromString(subId))
       }
-      goto(WaitingForSubWorkflowCost) using HasReceivedEventsData(target,
-                                                                  originalRequest,
-                                                                  metadataResponse.query,
-                                                                  metadataResponse.eventList,
-                                                                  Map.empty,
-                                                                  subWorkflowIds.size
+      goto(WaitingForSubWorkflowCost) using HasReceivedCostEventsData(target,
+                                                                      originalRequest,
+                                                                      metadataResponse.query,
+                                                                      metadataResponse.eventList,
+                                                                      status,
+                                                                      Map.empty,
+                                                                      subWorkflowIds.size
       )
     }
   }
 
   def buildCostAndStop(id: WorkflowId,
-                       status: String,
+                       status: WorkflowState,
                        eventsList: Seq[MetadataEvent],
                        expandedValues: Map[String, JsValue],
                        target: ActorRef,
@@ -644,7 +671,7 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props,
     val resp = JsObject(
       Map(
         WorkflowMetadataKeys.Id -> JsString(id.toString),
-        WorkflowMetadataKeys.Status -> JsString(status),
+        WorkflowMetadataKeys.Status -> JsString(status.toString),
         "currency" -> JsString(DefaultCurrency.getCurrencyCode),
         "cost" -> JsNumber(callCost + subworkflowCost),
         "errors" -> JsArray(costErrors ++ subworkflowErrors)
diff --git a/services/src/test/scala/cromwell/services/database/DatabaseTestKit.scala b/services/src/test/scala/cromwell/services/database/DatabaseTestKit.scala
index 5ca41bc15bc..54cffaa9e84 100644
--- a/services/src/test/scala/cromwell/services/database/DatabaseTestKit.scala
+++ b/services/src/test/scala/cromwell/services/database/DatabaseTestKit.scala
@@ -127,14 +127,14 @@ object DatabaseTestKit extends StrictLogging {
               )
             )
           case MysqlDatabasePlatform =>
-            Option(
-              MySQLContainer(
-                mysqlImageVersion = DockerImageName.parse(s"mysql:${networkDbSystem.dockerImageVersion}"),
-                databaseName = "cromwell_test",
-                username = "cromwell",
-                password = "test"
-              )
+            val mysqlContainer = MySQLContainer(
+              mysqlImageVersion = DockerImageName.parse(s"mysql:${networkDbSystem.dockerImageVersion}"),
+              databaseName = "cromwell_test",
+              username = "cromwell",
+              password = "test"
             )
+            mysqlContainer.container.withCommand("--innodb-print-all-deadlocks=ON")
+            Option(mysqlContainer)
           case PostgresqlDatabasePlatform =>
             Option(
               PostgreSQLContainer(
diff --git a/services/src/test/scala/cromwell/services/database/GroupMetricsSlickDatabaseSpec.scala b/services/src/test/scala/cromwell/services/database/GroupMetricsSlickDatabaseSpec.scala
index 53b7f5f001b..e70b0b12513 100644
--- a/services/src/test/scala/cromwell/services/database/GroupMetricsSlickDatabaseSpec.scala
+++ b/services/src/test/scala/cromwell/services/database/GroupMetricsSlickDatabaseSpec.scala
@@ -4,13 +4,13 @@ import com.dimafeng.testcontainers.Container
 import cromwell.core.Tags.DbmsTest
 import cromwell.database.sql.SqlConverters.OffsetDateTimeToSystemTimestamp
 import cromwell.database.sql.tables.GroupMetricsEntry
+import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
-import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.time.{Millis, Seconds, Span}
 
 import java.time.OffsetDateTime
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ExecutionContext, Future}
 
 class GroupMetricsSlickDatabaseSpec extends AnyFlatSpec with Matchers with ScalaFutures {
 
@@ -109,6 +109,34 @@ class GroupMetricsSlickDatabaseSpec extends AnyFlatSpec with Matchers with Scala
       } yield ()).futureValue
     }
 
+    it should "handle parallel upserts to database" taggedAs DbmsTest in {
+      val group1 = "groot-group-1"
+      val group2 = "groot-group-2"
+      val group3 = "rocket-group-1"
+      val group4 = "rocket-group-2"
+
+      // simulate parallel upsert for 4 different hog groups
+      // 4 * 25 => 100 parallel upserts
+      val inputGroups = (0 until 25).flatMap(_ => List(group1, group2, group3, group4))
+
+      (for {
+        _ <- Future.sequence(
+          inputGroups.map(r =>
+            dataAccess.recordGroupMetricsEntry(GroupMetricsEntry(r, OffsetDateTime.now.toSystemTimestamp))
+          )
+        )
+        rowCountGroup1 <- dataAccess.countGroupMetricsEntries(group1)
+        rowCountGroup2 <- dataAccess.countGroupMetricsEntries(group2)
+        rowCountGroup3 <- dataAccess.countGroupMetricsEntries(group3)
+        rowCountGroup4 <- dataAccess.countGroupMetricsEntries(group4)
+        // only 1 row for each hog group should exist in table
+        _ = rowCountGroup1 shouldBe 1
+        _ = rowCountGroup2 shouldBe 1
+        _ = rowCountGroup3 shouldBe 1
+        _ = rowCountGroup4 shouldBe 1
+      } yield ()).futureValue
+    }
+
     it should "close the database" taggedAs DbmsTest in {
       dataAccess.close()
     }
diff --git a/src/ci/bin/test.inc.sh b/src/ci/bin/test.inc.sh
index 6fb0f3465a0..ba2746b5031 100755
--- a/src/ci/bin/test.inc.sh
+++ b/src/ci/bin/test.inc.sh
@@ -543,7 +543,9 @@ cromwell::private::pip_install() {
 
 cromwell::private::upgrade_pip() {
     sudo apt-get install -y python3-pip
-    cromwell::private::pip_install pip --upgrade
+    # as of ubuntu 23 need to pass --user flag
+    # https://mail.openvswitch.org/pipermail/ovs-dev/2024-June/414969.html
+    cromwell::private::pip_install pip --upgrade --user pip
     cromwell::private::pip_install requests[security] --ignore-installed
 }
 
diff --git a/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/api/request/BatchRequestExecutor.scala b/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/api/request/BatchRequestExecutor.scala
index 547dbc853a8..38ebc66cf49 100644
--- a/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/api/request/BatchRequestExecutor.scala
+++ b/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/api/request/BatchRequestExecutor.scala
@@ -176,7 +176,7 @@ object BatchRequestExecutor {
     private def getEventList(events: List[StatusEvent]): List[ExecutionEvent] = {
       val startedRegex = ".*SCHEDULED to RUNNING.*".r
       val endedRegex = ".*RUNNING to.*".r // can be SUCCEEDED or FAILED
-      events.map { e =>
+      events.flatMap { e =>
         val time = java.time.Instant
           .ofEpochSecond(e.getEventTime.getSeconds, e.getEventTime.getNanos.toLong)
           .atOffset(java.time.ZoneOffset.UTC)
@@ -185,7 +185,14 @@ object BatchRequestExecutor {
           case endedRegex() => CallMetadataKeys.VmEndTime
           case _ => e.getType
         }
-        ExecutionEvent(name = eventType, offsetDateTime = time)
+        val executionEvents = List(ExecutionEvent(name = eventType, offsetDateTime = time))
+
+        // Add an additional ExecutionEvent to capture other info if the event is a VmStartTime or VmEndTime
+        if (eventType == CallMetadataKeys.VmStartTime || eventType == CallMetadataKeys.VmEndTime) {
+          executionEvents :+ ExecutionEvent(name = e.getDescription, offsetDateTime = time)
+        } else {
+          executionEvents
+        }
       }
     }
   }
diff --git a/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/models/GcpBatchRuntimeAttributes.scala b/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/models/GcpBatchRuntimeAttributes.scala
index 5c35d533257..7e908dd92af 100644
--- a/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/models/GcpBatchRuntimeAttributes.scala
+++ b/supportedBackends/google/batch/src/main/scala/cromwell/backend/google/batch/models/GcpBatchRuntimeAttributes.scala
@@ -60,7 +60,7 @@ object GcpBatchRuntimeAttributes {
 
   val BootDiskSizeKey = "bootDiskSizeGb"
   private val bootDiskValidationInstance = new IntRuntimeAttributesValidation(BootDiskSizeKey)
-  private val BootDiskDefaultValue = WomInteger(20)
+  private val BootDiskDefaultValue = WomInteger(10)
 
   val NoAddressKey = "noAddress"
   private val noAddressValidationInstance = new BooleanRuntimeAttributesValidation(NoAddressKey)
diff --git a/supportedBackends/google/batch/src/test/scala/cromwell/backend/google/batch/api/BatchRequestExecutorSpec.scala b/supportedBackends/google/batch/src/test/scala/cromwell/backend/google/batch/api/BatchRequestExecutorSpec.scala
index 79ac68772dd..4ce4c677645 100644
--- a/supportedBackends/google/batch/src/test/scala/cromwell/backend/google/batch/api/BatchRequestExecutorSpec.scala
+++ b/supportedBackends/google/batch/src/test/scala/cromwell/backend/google/batch/api/BatchRequestExecutorSpec.scala
@@ -161,13 +161,13 @@ class BatchRequestExecutorSpec
       case RunStatus.Success(events, _) =>
         val eventNames = events.map(_.name)
         val eventTimes = events.map(_.offsetDateTime.toString)
-        eventNames should contain theSameElementsAs List("vmStartTime", "vmEndTime")
-        eventTimes should contain theSameElementsAs List("1970-01-01T00:00:01Z", "1970-01-01T00:00:02Z")
+        eventNames should contain allOf ("vmStartTime", "vmEndTime")
+        eventTimes should contain allOf ("1970-01-01T00:00:01Z", "1970-01-01T00:00:02Z")
       case _ => fail("Expected RunStatus.Success with events")
     }
   }
 
-  it should "send vmStartTime and vmEndTime metadata info when a workflow fails" in {
+  it should "send vmStartTime and vmEndTime metadata info along with other events when a workflow fails" in {
     val mockClient = setupBatchClient(jobState = JobStatus.State.FAILED)
 
     // Create the BatchRequestExecutor
@@ -182,8 +182,10 @@ class BatchRequestExecutorSpec
       case RunStatus.Failed(_, events, _) =>
         val eventNames = events.map(_.name)
         val eventTimes = events.map(_.offsetDateTime.toString)
-        eventNames should contain theSameElementsAs List("vmStartTime", "vmEndTime")
-        eventTimes should contain theSameElementsAs List("1970-01-01T00:00:01Z", "1970-01-01T00:00:02Z")
+        println(eventNames)
+        eventNames should contain allOf ("vmStartTime", "vmEndTime")
+        eventNames should contain allOf ("Job state is set from RUNNING to SOME_OTHER_STATUS for job...", "Job state is set from SCHEDULED to RUNNING for job...")
+        eventTimes should contain allOf ("1970-01-01T00:00:01Z", "1970-01-01T00:00:02Z")
       case _ => fail("Expected RunStatus.Success with events")
     }
   }
diff --git a/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/PipelinesApiRuntimeAttributes.scala b/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/PipelinesApiRuntimeAttributes.scala
index bc05b525f5d..046840b6024 100644
--- a/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/PipelinesApiRuntimeAttributes.scala
+++ b/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/PipelinesApiRuntimeAttributes.scala
@@ -66,7 +66,7 @@ object PipelinesApiRuntimeAttributes {
 
   val BootDiskSizeKey = "bootDiskSizeGb"
   private val bootDiskValidationInstance = new IntRuntimeAttributesValidation(BootDiskSizeKey)
-  private val BootDiskDefaultValue = WomInteger(20)
+  private val BootDiskDefaultValue = WomInteger(10)
 
   val NoAddressKey = "noAddress"
   private val noAddressValidationInstance = new BooleanRuntimeAttributesValidation(NoAddressKey)
diff --git a/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/authentication/PipelinesApiVMAuthentication.scala b/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/authentication/PipelinesApiVMAuthentication.scala
index c00e2a6c16a..779dbbb855d 100644
--- a/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/authentication/PipelinesApiVMAuthentication.scala
+++ b/supportedBackends/google/pipelines/common/src/main/scala/cromwell/backend/google/pipelines/common/authentication/PipelinesApiVMAuthentication.scala
@@ -6,18 +6,6 @@ import common.validation.ErrorOr._
 import common.validation.Validation._
 import cromwell.cloudsupport.gcp.GoogleConfiguration
 import cromwell.core.DockerCredentials
-import spray.json.{JsString, JsValue}
-
-/**
- * Interface for Authentication information that can be included as a json object in the file uploaded to GCS
- * upon workflow creation and used in the VM.
- */
-sealed trait PipelinesApiAuthObject {
-  def context: String
-  def map: Map[String, JsValue]
-
-  def toMap: Map[String, Map[String, JsValue]] = Map(context -> map)
-}
 
 object PipelinesApiDockerCredentials {
 
@@ -53,10 +41,3 @@ case class PipelinesApiDockerCredentials(override val token: String,
                                          override val keyName: Option[String],
                                          override val authName: Option[String]
 ) extends DockerCredentials(token = token, keyName = keyName, authName = authName)
-    with PipelinesApiAuthObject {
-
-  override val context = "docker"
-  override val map = Map(
-    "token" -> JsString(token)
-  )
-}
diff --git a/supportedBackends/google/pipelines/v2beta/src/main/scala/cromwell/backend/google/pipelines/v2beta/LifeSciencesFactory.scala b/supportedBackends/google/pipelines/v2beta/src/main/scala/cromwell/backend/google/pipelines/v2beta/LifeSciencesFactory.scala
index 0d4996f63cc..bcb373a9026 100644
--- a/supportedBackends/google/pipelines/v2beta/src/main/scala/cromwell/backend/google/pipelines/v2beta/LifeSciencesFactory.scala
+++ b/supportedBackends/google/pipelines/v2beta/src/main/scala/cromwell/backend/google/pipelines/v2beta/LifeSciencesFactory.scala
@@ -154,13 +154,17 @@ case class LifeSciencesFactory(applicationName: String, authMode: GoogleAuthMode
        */
       val adjustedBootDiskSize = {
         val fromRuntimeAttributes = createPipelineParameters.runtimeAttributes.bootDiskSize
-        // Compute the decompressed size based on the information available
-        val userCommandImageSizeInBytes = createPipelineParameters.jobDescriptor.dockerSize
+
+        // Compute the decompressed size based on the information available. If we couldn't get the image size,
+        // default to 30GB. Defaulting to 0 can cause task to run out of disk. (more in AN-300)
+        val maybeUserCommandImageSizeInGB = createPipelineParameters.jobDescriptor.dockerSize
           .map(_.toFullSize(DockerConfiguration.instance.sizeCompressionFactor))
-          .getOrElse(0L)
-        val userCommandImageSizeInGB =
-          MemorySize(userCommandImageSizeInBytes.toDouble, MemoryUnit.Bytes).to(MemoryUnit.GB).amount
-        val userCommandImageSizeRoundedUpInGB = userCommandImageSizeInGB.ceil.toInt
+          .map(s => MemorySize(s.toDouble, MemoryUnit.Bytes).to(MemoryUnit.GB))
+        val (userCommandImageSizeInGB, userImageLogString) = maybeUserCommandImageSizeInGB match {
+          case Some(imageSize) => (imageSize, "user command image")
+          case None => (MemorySize(30, MemoryUnit.GB), "failed to obtain user command image size, using safe default")
+        }
+        val userCommandImageSizeRoundedUpInGB = userCommandImageSizeInGB.amount.ceil.toInt
 
         val totalSize = fromRuntimeAttributes +
           createPipelineParameters.dockerImageCacheDiskOpt
@@ -171,7 +175,7 @@ case class LifeSciencesFactory(applicationName: String, authMode: GoogleAuthMode
 
         if (totalSize != fromRuntimeAttributes) {
           jobLogger.info(
-            s"Adjusting boot disk size to $totalSize GB: $fromRuntimeAttributes GB (runtime attributes) + $userCommandImageSizeRoundedUpInGB GB (user command image) + ${ActionUtils.cromwellImagesSizeRoundedUpInGB} GB (Cromwell support images)"
+            s"Adjusting boot disk size to $totalSize GB: $fromRuntimeAttributes GB (runtime attributes) + $userCommandImageSizeRoundedUpInGB GB ($userImageLogString) + ${ActionUtils.cromwellImagesSizeRoundedUpInGB} GB (Cromwell support images)"
           )
         }