Merge pull request #3265 from magda-io/cherry-pick-3251

Cherry-pick PR for issue #3251 from next branch into v1.1.0

CHANGES.md

@@ -9,6 +9,7 @@
 - #3263 Build Multi-Arch (`linux/amd64` & `linux/arm64`) Docker Images in CI (Except `magda-postgres` & `magda-elasticsearch`)
 - Related to #3263, Build Multi-Arch (`linux/amd64` & `linux/arm64`) Docker Image for `magda-elasticsearch` as well.
 - Related to #3263, adjusted helm chart for elasticsearch to make it run properly on linux/arm64 platform.
+- #3251 Fixed akka HTTP client POST request racing conditions
 
 ## 1.0.0
 

magda-int-test/src/test/scala/au/csiro/data61/magda/indexer/RegisterWebhookSpec.scala

@@ -25,15 +25,16 @@ class RegisterWebhookSpec extends BaseRegistryApiSpec with SprayJsonSupport {
 
       // Expect the new hook to be posted
       (param.authFetcher
-        .post(_: String, _: WebHook, _: Seq[HttpHeader])(
+        .post(_: String, _: WebHook, _: Seq[HttpHeader], _: Boolean)(
           _: ToEntityMarshaller[WebHook]
         ))
-        .expects(s"/v0/hooks", *, *, *)
+        .expects(s"/v0/hooks", *, *, *, *)
         .onCall(
           (
               url: String,
               webhook: WebHook,
               headers: Seq[HttpHeader],
+              autoRetry: Boolean,
               marshaller: ToEntityMarshaller[WebHook]
           ) => {
             // Forward the req to the registry api
@@ -113,15 +114,16 @@ class RegisterWebhookSpec extends BaseRegistryApiSpec with SprayJsonSupport {
 
       // Expect the hook to update itself
       (param.authFetcher
-        .put(_: String, _: WebHook, _: Seq[HttpHeader])(
+        .put(_: String, _: WebHook, _: Seq[HttpHeader], _: Boolean)(
           _: ToEntityMarshaller[WebHook]
         ))
-        .expects(s"/v0/hooks/$webhookId", *, *, *)
+        .expects(s"/v0/hooks/$webhookId", *, *, *, *)
         .onCall(
           (
               url: String,
               webhook: WebHook,
               headers: Seq[HttpHeader],
+              autoRetry: Boolean,
               marshaller: ToEntityMarshaller[WebHook]
           ) => {
             // Forward the req to the registry api
@@ -143,15 +145,21 @@ class RegisterWebhookSpec extends BaseRegistryApiSpec with SprayJsonSupport {
 
       // Expect an ACK call once the indexer has determined that the webhook already exists
       (param.authFetcher
-        .post(_: String, _: WebHookAcknowledgement, _: Seq[HttpHeader])(
+        .post(
+          _: String,
+          _: WebHookAcknowledgement,
+          _: Seq[HttpHeader],
+          _: Boolean
+        )(
           _: ToEntityMarshaller[WebHookAcknowledgement]
         ))
-        .expects(s"/v0/hooks/$webhookId/ack", *, *, *)
+        .expects(s"/v0/hooks/$webhookId/ack", *, *, *, *)
         .onCall(
           (
               url: String,
               webhookAck: WebHookAcknowledgement,
               headers: Seq[HttpHeader],
+              autoRetry: Boolean,
               marshaller: ToEntityMarshaller[WebHookAcknowledgement]
           ) => {
             // Don't forward this to the registry, just check the info passed is right.

magda-registry-api/src/test/scala/au/csiro/data61/magda/registry/BaseRecordsServiceAuthSpec.scala

@@ -2583,7 +2583,8 @@ abstract class BaseRecordsServiceAuthSpec extends ApiSpec {
       .post(
         _: String,
         _: HttpEntity.Strict,
-        _: List[HttpHeader]
+        _: List[HttpHeader],
+        _: Boolean
       )(
         _: ToEntityMarshaller[HttpEntity.Strict]
       ))
@@ -2597,6 +2598,7 @@ abstract class BaseRecordsServiceAuthSpec extends ApiSpec {
              |}""".stripMargin
         ),
         *,
+        true,
         *
       )
       .returning(

magda-scala-common/src/main/scala/au/csiro/data61/magda/client/AuthApiClient.scala

@@ -94,7 +94,8 @@ class AuthApiClient(authHttpFetcher: HttpFetcher)(
       .post(
         s"/v0/opa/compile",
         HttpEntity(ContentTypes.`application/json`, requestData),
-        headers
+        headers,
+        true
       )
       .flatMap(receiveOpaResponse[List[List[OpaQuery]]](_, policyId) { json =>
         OpaParser.parseOpaResponse(json, policyId)

magda-scala-common/src/main/scala/au/csiro/data61/magda/client/HttpFetcher.scala

@@ -20,11 +20,21 @@ import akka.http.scaladsl.model.HttpHeader
 trait HttpFetcher {
   def get(path: String, headers: Seq[HttpHeader] = Seq()): Future[HttpResponse]
 
-  def post[T](path: String, payload: T, headers: Seq[HttpHeader] = Seq())(
+  def post[T](
+      path: String,
+      payload: T,
+      headers: Seq[HttpHeader] = Seq(),
+      autoRetryConnection: Boolean = false
+  )(
       implicit m: ToEntityMarshaller[T]
   ): Future[HttpResponse]
 
-  def put[T](path: String, payload: T, headers: Seq[HttpHeader] = Seq())(
+  def put[T](
+      path: String,
+      payload: T,
+      headers: Seq[HttpHeader] = Seq(),
+      autoRetryConnection: Boolean = false
+  )(
       implicit m: ToEntityMarshaller[T]
   ): Future[HttpResponse]
 }
@@ -59,7 +69,12 @@ class HttpFetcherImpl(baseUrl: URL)(
     }
   }
 
-  def post[T](path: String, payload: T, headers: Seq[HttpHeader] = Seq())(
+  def post[T](
+      path: String,
+      payload: T,
+      headers: Seq[HttpHeader] = Seq(),
+      autoRetryConnection: Boolean = false
+  )(
       implicit m: ToEntityMarshaller[T]
   ): Future[HttpResponse] = {
     val url = s"${baseUrl.getPath}${path}"
@@ -72,12 +87,41 @@ class HttpFetcherImpl(baseUrl: URL)(
     val request = RequestBuilding
       .Post(url, payload)
       .withHeaders(scala.collection.immutable.Seq.concat(headers))
-    Source.single((request, 0)).via(connectionFlow).runWith(Sink.head).map {
-      case (response, _) => response.get
+    val result =
+      Source.single((request, 0)).via(connectionFlow).runWith(Sink.head).map {
+        case (response, _) => response.get
+      }
+
+    if (!autoRetryConnection) {
+      result
+    } else {
+      // Retry the request for racing condition
+      // https://github.com/magda-io/magda/issues/3251
+      result.recoverWith {
+        case error: Throwable
+            // We have to do this as UnexpectedConnectionClosureException is a private class
+            // https://github.com/akka/akka-http/issues/768
+            if error.getMessage.contains(
+              "The http server closed the connection unexpectedly"
+            ) =>
+          Source
+            .single((request, 0))
+            .via(connectionFlow)
+            .runWith(Sink.head)
+            .map {
+              case (response, _) => response.get
+            }
+      }
     }
+
   }
 
-  def put[T](path: String, payload: T, headers: Seq[HttpHeader] = Seq())(
+  def put[T](
+      path: String,
+      payload: T,
+      headers: Seq[HttpHeader] = Seq(),
+      autoRetryConnection: Boolean = false
+  )(
       implicit m: ToEntityMarshaller[T]
   ): Future[HttpResponse] = {
     val url = s"${baseUrl.getPath}${path}"
@@ -90,8 +134,31 @@ class HttpFetcherImpl(baseUrl: URL)(
     val request = RequestBuilding
       .Put(url, payload)
       .withHeaders(scala.collection.immutable.Seq.concat(headers))
-    Source.single((request, 0)).via(connectionFlow).runWith(Sink.head).map {
-      case (response, _) => response.get
+    val result =
+      Source.single((request, 0)).via(connectionFlow).runWith(Sink.head).map {
+        case (response, _) => response.get
+      }
+
+    if (!autoRetryConnection) {
+      result
+    } else {
+      // Retry the request for racing condition
+      // https://github.com/magda-io/magda/issues/3251
+      result.recoverWith {
+        case error: Throwable
+            // We have to do this as UnexpectedConnectionClosureException is a private class
+            // https://github.com/akka/akka-http/issues/768
+            if error.getMessage.contains(
+              "The http server closed the connection unexpectedly"
+            ) =>
+          Source
+            .single((request, 0))
+            .via(connectionFlow)
+            .runWith(Sink.head)
+            .map {
+              case (response, _) => response.get
+            }
+      }
     }
   }