From 976ca400eb038a1bdf66c7ab6412455a03c3e64d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Victor=20Peter=20Rouven=20M=C3=BCller?= Date: Fri, 10 Sep 2021 10:45:58 +0200 Subject: [PATCH] Force JSON API to refresh packages for GET requests to /v1/query (#10835) * Add failing test that covers the bug we found in #10823 * Fix /v1/query endpoint bug changelog_begin - [JSON API] Fixed a bug that prevented the JSON API to be aware of packages uploaded directly via the Ledger API. changelog_end --- .../AbstractHttpServiceIntegrationTest.scala | 37 +++++++++++++++++++ .../digitalasset/http/ContractsService.scala | 11 ++++-- .../digitalasset/http/PackageService.scala | 17 +++++++-- 3 files changed, 58 insertions(+), 7 deletions(-) diff --git a/ledger-service/http-json/src/itlib/scala/http/AbstractHttpServiceIntegrationTest.scala b/ledger-service/http-json/src/itlib/scala/http/AbstractHttpServiceIntegrationTest.scala index 945bb7177c03..ff822b33d2f7 100644 --- a/ledger-service/http-json/src/itlib/scala/http/AbstractHttpServiceIntegrationTest.scala +++ b/ledger-service/http-json/src/itlib/scala/http/AbstractHttpServiceIntegrationTest.scala @@ -50,6 +50,7 @@ import scalaz.syntax.apply._ import scala.concurrent.{ExecutionContext, Future} import scala.util.{Success, Try} import com.daml.ledger.api.{domain => LedgerApiDomain} +import com.daml.ports.Port object AbstractHttpServiceIntegrationTestFuns { private[http] val dar1 = requiredResource("docs/quickstart-model.dar") @@ -151,11 +152,26 @@ trait AbstractHttpServiceIntegrationTestFuns extends StrictLogging { ): Future[A] = withHttpServiceAndClient((a, b, c, _, ledgerId) => f(a, b, c, ledgerId)) + protected def withHttpServiceOnly[A](ledgerPort: Port)( + f: (Uri, DomainJsonEncoder, DomainJsonDecoder) => Future[A] + ): Future[A] = + HttpServiceTestFixture.withHttpService[A]( + testId, + ledgerPort, + jdbcConfig, + staticContentConfig, + useTls = useTls, + wsConfig = wsConfig, + )((uri, encoder, decoder, _) => f(uri, encoder, decoder)) + protected def withLedger[A](testFn: (DamlLedgerClient, LedgerId) => Future[A]): Future[A] = HttpServiceTestFixture.withLedger[A](List(dar1, dar2), testId) { case (_, client, ledgerId) => testFn(client, ledgerId) } + protected def withLedger2[A](testFn: (Port, DamlLedgerClient, LedgerId) => Future[A]): Future[A] = + HttpServiceTestFixture.withLedger[A](List(dar1, dar2), testId)(testFn) + protected val headersWithAuth = authorizationHeader(jwt) protected def headersWithPartyAuth(actAs: List[String], readAs: List[String] = List()) = @@ -1618,6 +1634,27 @@ abstract class AbstractHttpServiceIntegrationTest }: Future[Assertion] } + "package list is updated when a query request is made" in withLedger2[Assertion] { + (ledgerPort: Port, _, _) => + for { + _ <- withHttpServiceOnly(ledgerPort) { (uri, encoder, _) => + searchDataSet.traverse(c => postCreateCommand(c, encoder, uri)).flatMap { rs => + rs.map(_._1) shouldBe List.fill(searchDataSet.size)(StatusCodes.OK) + } + } + _ <- withHttpServiceOnly(ledgerPort) { (uri, _, _) => + getRequest(uri = uri.withPath(Uri.Path("/v1/query"))) + .flatMap { case (status, output) => + status shouldBe StatusCodes.OK + assertStatus(output, StatusCodes.OK) + inside(getResult(output)) { case JsArray(result) => + result should have length 4 + } + }: Future[Assertion] + } + } yield succeed + } + "archiving a large number of contracts should succeed" in withHttpServiceAndClient( StartSettings.DefaultMaxInboundMessageSize * 10 ) { (uri, encoder, _, _, _) => diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala index 89377f637007..8a3149614d3d 100644 --- a/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala +++ b/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala @@ -198,7 +198,7 @@ class ContractsService( x => resolveTemplateId(lc)(jwt, ledgerId)(x) .map(_.toOption.flatten.map(Set(_))), - Future.successful(allTemplateIds().some), + allTemplateIds(lc)(jwt, ledgerId).map(_.some), ) ) @@ -250,9 +250,12 @@ class ContractsService( lc: LoggingContextOf[InstanceUUID] ): SearchResult[Error \/ domain.ActiveContract[LfValue]] = domain.OkResponse( - Source(allTemplateIds()).flatMapConcat(x => - searchInMemoryOneTpId(jwt, ledgerId, parties, x, _ => true) - ) + Source + .future(allTemplateIds(lc)(jwt, ledgerId)) + .flatMapConcat(x => + Source(x) + .flatMapConcat(x => searchInMemoryOneTpId(jwt, ledgerId, parties, x, _ => true)) + ) ) def search( diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/PackageService.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/PackageService.scala index 293dfe48e721..a8846b75f7f2 100644 --- a/ledger-service/http-json/src/main/scala/com/digitalasset/http/PackageService.scala +++ b/ledger-service/http-json/src/main/scala/com/digitalasset/http/PackageService.scala @@ -176,8 +176,17 @@ private class PackageService( .TypeCon(iface.TypeConName(IdentifierConverters.lfIdentifier(templateId)), ImmArraySeq()) ) - def allTemplateIds: AllTemplateIds = - () => state.templateIdMap.all + def allTemplateIds(implicit ec: ExecutionContext): AllTemplateIds = { + implicit lc => (jwt, ledgerId) => + val f = + if (cache.packagesShouldBeFetchedAgain) { + logger.trace( + "no package id and we do have the package, refresh because of timeout" + ) + reload(jwt, ledgerId) + } else Future.successful(()) + f.map(_ => state.templateIdMap.all) + } // See the above comment def resolveChoiceArgType: ResolveChoiceArgType = @@ -214,7 +223,9 @@ object PackageService { TemplateId.RequiredPkg => Error \/ iface.Type type AllTemplateIds = - () => Set[TemplateId.RequiredPkg] + LoggingContextOf[ + InstanceUUID + ] => (Jwt, LedgerApiDomain.LedgerId) => Future[Set[TemplateId.RequiredPkg]] type ResolveChoiceArgType = (TemplateId.RequiredPkg, Choice) => Error \/ iface.Type