Use fetchAndPersistBracket instead of fetchAndPersist (#12672)

* Use fetchAndPersistBracket instead of fetchAndPersist

* add changelog

CHANGELOG_BEGIN
- [JSON API] Under rare conditions, a POST-style multi-template query backed by
  database could have different template IDs appear to be queried at different
  ledger times, if updated concurrently.  This conditions is now checked and
  accounted for.  See `issue #12672
  <https://github.com/digital-asset/daml/pull/12672>`__.
CHANGELOG_END

* add a searchQuery metric event for each fetch trial in sync search

* revert the contract ID and contract key fetches, and explain why

Co-authored-by: Stephen Compall <stephen.compall@daml.com>

ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsFetch.scala

@@ -38,7 +38,7 @@ import scalaz.syntax.functor._
 import scalaz.syntax.foldable._
 import scalaz.syntax.order._
 import scalaz.syntax.std.option._
-import scalaz.\/
+import scalaz.{~>, \/, NaturalTransformation}
 import spray.json.{JsNull, JsValue}
 
 import scala.concurrent.ExecutionContext
@@ -65,6 +65,7 @@ private class ContractsFetch(
       ledgerId: LedgerApiDomain.LedgerId,
       parties: domain.PartySet,
       templateIds: List[domain.TemplateId.RequiredPkg],
+      tickFetch: ConnectionIO ~> ConnectionIO = NaturalTransformation.refl,
   )(within: BeginBookmark[Terminates.AtAbsolute] => ConnectionIO[A])(implicit
       ec: ExecutionContext,
       mat: Materializer,
@@ -78,7 +79,7 @@ private class ContractsFetch(
         fetchTemplateIds: List[domain.TemplateId.RequiredPkg],
         absEnd: Terminates.AtAbsolute,
     ): ConnectionIO[A] = for {
-      bb <- fetchToAbsEnd(fetchContext, fetchTemplateIds, absEnd)
+      bb <- tickFetch(fetchToAbsEnd(fetchContext, fetchTemplateIds, absEnd))
       a <- within(bb)
       // fetchTemplateIds can be a subset of templateIds (or even empty),
       // but we only get away with that by checking _all_ of templateIds,

ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala

@@ -31,7 +31,7 @@ import scalaz.std.option._
 import scalaz.syntax.show._
 import scalaz.syntax.std.option._
 import scalaz.syntax.traverse._
-import scalaz.{-\/, OneAnd, OptionT, Show, \/, \/-}
+import scalaz.{~>, -\/, OneAnd, OptionT, Show, \/, \/-}
 import spray.json.JsValue
 
 import scala.concurrent.{ExecutionContext, Future}
@@ -40,6 +40,7 @@ import scalaz.std.scalaFuture._
 
 import com.codahale.metrics.Timer
 import doobie.free.{connection => fconn}
+import fconn.ConnectionIO
 
 class ContractsService(
     resolveTemplateId: PackageService.ResolveTemplateId,
@@ -326,6 +327,9 @@ class ContractsService(
             )
             res <- OptionT(unsafeRunAsync {
               import doobie.implicits._, cats.syntax.apply._
+              // a single contractId is either present or not; we would only need
+              // to fetchAndPersistBracket if we were looking up multiple cids
+              // in the same HTTP request, and they would all have to be bracketed once -SC
               timed(
                 metrics.daml.HttpJsonApi.Db.fetchByIdFetch,
                 fetch.fetchAndPersist(jwt, ledgerId, parties, List(resolved)),
@@ -354,6 +358,13 @@ class ContractsService(
             resolved <- resolveTemplateId(lc)(jwt, ledgerId)(templateId).map(_.toOption.flatten.get)
             found <- unsafeRunAsync {
               import doobie.implicits._, cats.syntax.apply._
+              // it is possible for the contract under a given key to have been
+              // replaced concurrently with a contract unobservable by parties, i.e.
+              // the DB contains two contracts with the same key.  However, this doesn't
+              // ever yield an _inconsistent_ result, merely one that is slightly back-in-time,
+              // which is true of all json-api responses.  Again, if we were checking for
+              // multiple template/contract-key pairs in a single request, they would all
+              // have to be contained within a single fetchAndPersistBracket -SC
               timed(
                 metrics.daml.HttpJsonApi.Db.fetchByKeyFetch,
                 fetch.fetchAndPersist(jwt, ledgerId, parties, List(resolved)),
@@ -412,15 +423,21 @@ class ContractsService(
           import doobie.implicits._
           import ctx.{jwt, parties, templateIds, ledgerId}
           for {
-            _ <- timed(
-              metrics.daml.HttpJsonApi.Db.searchFetch,
-              fetch.fetchAndPersist(jwt, ledgerId, parties, templateIds.toList),
-            )
-            cts <- timed(
-              metrics.daml.HttpJsonApi.Db.searchQuery,
-              templateIds.toVector
-                .traverse(tpId => searchDbOneTpId_(parties, tpId, queryParams)),
-            )
+            cts <- fetch.fetchAndPersistBracket(
+              jwt,
+              ledgerId,
+              parties,
+              templateIds.toList,
+              Lambda[ConnectionIO ~> ConnectionIO](
+                timed(metrics.daml.HttpJsonApi.Db.searchFetch, _)
+              ),
+            ) { _ =>
+              timed(
+                metrics.daml.HttpJsonApi.Db.searchQuery,
+                templateIds.toVector
+                  .traverse(tpId => searchDbOneTpId_(parties, tpId, queryParams)),
+              )
+            }
           } yield cts.flatten
         }