Skip to content

Commit

Permalink
wip
Browse files Browse the repository at this point in the history
  • Loading branch information
@rautenrieth-da
rautenrieth-da committed Oct 8, 2021
1 parent 45235e3 commit 29d6d39
Show file tree
Hide file tree
Showing 19 changed files with 121 additions and 163 deletions.
3 changes: 1 addition & 2 deletions .../participant-integration-api/src/main/scala/platform/index/LedgerBackedIndexService.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@

package com.daml.platform.index

import java.time.Instant
import akka.NotUsed
import akka.stream.scaladsl.Source
import com.daml.daml_lf_dev.DamlLf.Archive
Expand Down Expand Up @@ -204,7 +203,7 @@ private[platform] final class LedgerBackedIndexService(

override def lookupMaximumLedgerTime(ids: Set[ContractId])(implicit
loggingContext: LoggingContext
): Future[Option[Instant]] =
): Future[Option[Timestamp]] =
ledger.lookupMaximumLedgerTime(ids)

override def lookupContractKey(
Expand Down
3 changes: 1 addition & 2 deletions ledger/participant-integration-api/src/main/scala/platform/index/MeteredReadOnlyLedger.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@

package com.daml.platform.index

import java.time.Instant
import akka.NotUsed
import akka.stream.scaladsl.Source
import com.daml.daml_lf_dev.DamlLf.Archive
Expand Down Expand Up @@ -104,7 +103,7 @@ private[platform] class MeteredReadOnlyLedger(ledger: ReadOnlyLedger, metrics: M

override def lookupMaximumLedgerTime(
contractIds: Set[ContractId]
)(implicit loggingContext: LoggingContext): Future[Option[Instant]] =
)(implicit loggingContext: LoggingContext): Future[Option[Timestamp]] =
Timed.future(
metrics.daml.index.lookupMaximumLedgerTime,
ledger.lookupMaximumLedgerTime(contractIds),
Expand Down
31 changes: 15 additions & 16 deletions ledger/participant-integration-api/src/main/scala/platform/indexer/ExecuteUpdate.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ trait ExecuteUpdate {
completionInfo = tx.optCompletionInfo,
workflowId = tx.transactionMeta.workflowId,
transactionId = tx.transactionId,
ledgerEffectiveTime = tx.transactionMeta.ledgerEffectiveTime.toInstant,
ledgerEffectiveTime = tx.transactionMeta.ledgerEffectiveTime,
offset = offsetStepPair.offsetStep.offset,
transaction = tx.transaction,
divulgedContracts = tx.divulgedContracts,
Expand Down Expand Up @@ -129,7 +129,7 @@ trait ExecuteUpdate {
) =>
val entry = PartyLedgerEntry.AllocationAccepted(
submissionId,
recordTime.toInstant,
recordTime,
domain.PartyDetails(party, Some(displayName), participantId == hostingParticipantId),
)
ledgerDao.storePartyEntry(offsetStep, entry)
Expand All @@ -142,38 +142,37 @@ trait ExecuteUpdate {
) =>
val entry = PartyLedgerEntry.AllocationRejected(
submissionId,
recordTime.toInstant,
recordTime,
rejectionReason,
)
ledgerDao.storePartyEntry(offsetStep, entry)

case PublicPackageUpload(archives, optSourceDescription, recordTime, optSubmissionId) =>
val recordTimeInstant = recordTime.toInstant
val packages: List[(DamlLf.Archive, v2.PackageDetails)] = archives.map(archive =>
archive -> v2.PackageDetails(
size = archive.getPayload.size.toLong,
knownSince = recordTimeInstant,
knownSince = recordTime,
sourceDescription = optSourceDescription,
)
)
val optEntry: Option[PackageLedgerEntry] =
optSubmissionId.map(submissionId =>
PackageLedgerEntry.PackageUploadAccepted(submissionId, recordTimeInstant)
PackageLedgerEntry.PackageUploadAccepted(submissionId, recordTime)
)
ledgerDao.storePackageEntry(offsetStep, packages, optEntry)

case PublicPackageUploadRejected(submissionId, recordTime, rejectionReason) =>
val entry = PackageLedgerEntry.PackageUploadRejected(
submissionId,
recordTime.toInstant,
recordTime,
rejectionReason,
)
ledgerDao.storePackageEntry(offsetStep, List.empty, Some(entry))

case config: ConfigurationChanged =>
ledgerDao.storeConfigurationEntry(
offsetStep,
config.recordTime.toInstant,
config.recordTime,
config.submissionId,
config.newConfiguration,
None,
Expand All @@ -182,7 +181,7 @@ trait ExecuteUpdate {
case configRejection: ConfigurationChangeRejected =>
ledgerDao.storeConfigurationEntry(
offsetStep,
configRejection.recordTime.toInstant,
configRejection.recordTime,
configRejection.submissionId,
configRejection.proposedConfiguration,
Some(configRejection.rejectionReason),
Expand All @@ -191,7 +190,7 @@ trait ExecuteUpdate {
case CommandRejected(recordTime, completionInfo, reason) =>
ledgerDao.storeRejection(
Some(completionInfo),
recordTime.toInstant,
recordTime,
offsetStep,
reason,
)
Expand All @@ -206,16 +205,16 @@ trait ExecuteUpdate {
completionInfo = optCompletionInfo,
workflowId = transactionMeta.workflowId,
transactionId = transactionId,
ledgerEffectiveTime = transactionMeta.ledgerEffectiveTime.toInstant,
ledgerEffectiveTime = transactionMeta.ledgerEffectiveTime,
offset = offsetStep.offset,
transaction = transaction,
divulgedContracts = divulgedContracts,
blindingInfo = blindingInfo,
),
completionInfo = optCompletionInfo,
transactionId = transactionId,
recordTime = recordTime.toInstant,
ledgerEffectiveTime = transactionMeta.ledgerEffectiveTime.toInstant,
recordTime = recordTime,
ledgerEffectiveTime = transactionMeta.ledgerEffectiveTime,
offsetStep = offsetStep,
transaction = transaction,
divulged = divulgedContracts,
Expand Down Expand Up @@ -291,7 +290,7 @@ class PipelinedExecuteUpdate(
ledgerDao.completeTransaction(
completionInfo = tx.optCompletionInfo,
transactionId = tx.transactionId,
recordTime = tx.recordTime.toInstant,
recordTime = tx.recordTime,
offsetStep = offsetStep,
),
)
Expand Down Expand Up @@ -383,8 +382,8 @@ class AtomicExecuteUpdate(
preparedInsert,
completionInfo = optCompletionInfo,
transactionId = transactionId,
recordTime = recordTime.toInstant,
ledgerEffectiveTime = transactionMeta.ledgerEffectiveTime.toInstant,
recordTime = recordTime,
ledgerEffectiveTime = transactionMeta.ledgerEffectiveTime,
offsetStep = offsetStep,
transaction = transaction,
divulged = divulgedContracts,
Expand Down
2 changes: 1 addition & 1 deletion ledger/participant-integration-api/src/main/scala/platform/store/BaseLedger.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ private[platform] abstract class BaseLedger(

override def lookupMaximumLedgerTime(
contractIds: Set[ContractId]
)(implicit loggingContext: LoggingContext): Future[Option[Instant]] =
)(implicit loggingContext: LoggingContext): Future[Option[Timestamp]] =
contractStore.lookupMaximumLedgerTime(contractIds)

override def getParties(parties: Seq[Ref.Party])(implicit
Expand Down
2 changes: 1 addition & 1 deletion ledger/participant-integration-api/src/main/scala/platform/store/ReadOnlyLedger.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ private[platform] trait ReadOnlyLedger extends ReportsHealth with AutoCloseable

def lookupMaximumLedgerTime(
contractIds: Set[ContractId]
)(implicit loggingContext: LoggingContext): Future[Option[Instant]]
)(implicit loggingContext: LoggingContext): Future[Option[Timestamp]]

def lookupKey(key: GlobalKey, forParties: Set[Ref.Party])(implicit
loggingContext: LoggingContext
Expand Down
57 changes: 27 additions & 30 deletions ...rticipant-integration-api/src/main/scala/platform/store/appendonlydao/JdbcLedgerDao.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ private class JdbcLedgerDao(

override def storeConfigurationEntry(
offsetStep: OffsetStep,
recordedAt: Instant,
recordedAt: Timestamp,
submissionId: String,
configuration: Configuration,
rejectionReason: Option[String],
Expand Down Expand Up @@ -191,7 +191,7 @@ private class JdbcLedgerDao(
val update = finalRejectionReason match {
case None =>
state.Update.ConfigurationChanged(
recordTime = Time.Timestamp.assertFromInstant(recordedAt),
recordTime = recordedAt,
submissionId = Ref.SubmissionId.assertFromString(submissionId),
participantId =
Ref.ParticipantId.assertFromString("1"), // not used for DbDto generation
Expand All @@ -200,7 +200,7 @@ private class JdbcLedgerDao(

case Some(reason) =>
state.Update.ConfigurationChangeRejected(
recordTime = Time.Timestamp.assertFromInstant(recordedAt),
recordTime = recordedAt,
submissionId = Ref.SubmissionId.assertFromString(submissionId),
participantId =
Ref.ParticipantId.assertFromString("1"), // not used for DbDto generation
Expand Down Expand Up @@ -240,7 +240,7 @@ private class JdbcLedgerDao(
//
// This will be properly resolved once we move away from the `sandbox-classic` codebase.
participantId = if (partyDetails.isLocal) participantId else NonLocalParticipantId,
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
submissionId = submissionIdOpt,
)
),
Expand All @@ -255,7 +255,7 @@ private class JdbcLedgerDao(
state.Update.PartyAllocationRejected(
submissionId = submissionId,
participantId = participantId,
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
rejectionReason = reason,
)
),
Expand Down Expand Up @@ -287,7 +287,7 @@ private class JdbcLedgerDao(
completionInfo: Option[state.CompletionInfo],
workflowId: Option[Ref.WorkflowId],
transactionId: Ref.TransactionId,
ledgerEffectiveTime: Instant,
ledgerEffectiveTime: Timestamp,
offset: Offset,
transaction: CommittedTransaction,
divulgedContracts: Iterable[state.DivulgedContract],
Expand Down Expand Up @@ -315,7 +315,7 @@ private class JdbcLedgerDao(
override def completeTransaction(
completionInfo: Option[state.CompletionInfo],
transactionId: Ref.TransactionId,
recordTime: Instant,
recordTime: Timestamp,
offsetStep: OffsetStep,
)(implicit loggingContext: LoggingContext): Future[PersistenceResponse] =
throw new UnsupportedOperationException(
Expand All @@ -326,8 +326,8 @@ private class JdbcLedgerDao(
preparedInsert: PreparedInsert,
completionInfo: Option[state.CompletionInfo],
transactionId: Ref.TransactionId,
recordTime: Instant,
ledgerEffectiveTime: Instant,
recordTime: Timestamp,
ledgerEffectiveTime: Timestamp,
offsetStep: OffsetStep,
transaction: CommittedTransaction,
divulged: Iterable[state.DivulgedContract],
Expand All @@ -337,7 +337,7 @@ private class JdbcLedgerDao(
) // TODO append-only: cleanup

private def validate(
ledgerEffectiveTime: Instant,
ledgerEffectiveTime: Timestamp,
transaction: CommittedTransaction,
divulged: Iterable[state.DivulgedContract],
)(implicit connection: Connection): Option[PostCommitValidation.Rejection] =
Expand All @@ -352,7 +352,7 @@ private class JdbcLedgerDao(

override def storeRejection(
completionInfo: Option[state.CompletionInfo],
recordTime: Instant,
recordTime: Timestamp,
offsetStep: OffsetStep,
reason: state.Update.CommandRejected.RejectionReasonTemplate,
)(implicit loggingContext: LoggingContext): Future[PersistenceResponse] =
Expand All @@ -364,7 +364,7 @@ private class JdbcLedgerDao(
offset,
completionInfo.map(info =>
state.Update.CommandRejected(
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
completionInfo = info,
reasonTemplate = reason,
)
Expand Down Expand Up @@ -403,8 +403,7 @@ private class JdbcLedgerDao(
state.Update.TransactionAccepted(
optCompletionInfo = completionInfo,
transactionMeta = state.TransactionMeta(
ledgerEffectiveTime =
Time.Timestamp.assertFromInstant(tx.ledgerEffectiveTime),
ledgerEffectiveTime = tx.ledgerEffectiveTime,
workflowId = tx.workflowId,
submissionTime = null, // not used for DbDto generation
submissionSeed = null, // not used for DbDto generation
Expand All @@ -414,7 +413,7 @@ private class JdbcLedgerDao(
),
transaction = tx.transaction,
transactionId = tx.transactionId,
recordTime = Time.Timestamp.assertFromInstant(tx.recordedAt),
recordTime = tx.recordedAt,
divulgedContracts = Nil,
blindingInfo = None,
)
Expand All @@ -433,7 +432,7 @@ private class JdbcLedgerDao(
offset,
Some(
state.Update.CommandRejected(
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
completionInfo = state
.CompletionInfo(actAs, applicationId, commandId, None, Some(submissionId)),
reasonTemplate = reason.toParticipantStateRejectionReason,
Expand Down Expand Up @@ -507,13 +506,11 @@ private class JdbcLedgerDao(
sourceDescription = packages.headOption.flatMap(
_._2.sourceDescription
),
recordTime = Time.Timestamp.assertFromInstant(
packages.headOption
.map(
_._2.knownSince
)
.getOrElse(Instant.EPOCH)
),
recordTime = packages.headOption
.map(
_._2.knownSince
)
.getOrElse(Timestamp.Epoch),
submissionId =
None, // If the submission ID is missing, this update will not insert a row in the package_entries table
)
Expand All @@ -524,14 +521,14 @@ private class JdbcLedgerDao(
sourceDescription = packages.headOption.flatMap(
_._2.sourceDescription
),
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
submissionId = Some(submissionId),
)

case Some(PackageLedgerEntry.PackageUploadRejected(submissionId, recordTime, reason)) =>
state.Update.PublicPackageUploadRejected(
submissionId = submissionId,
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
rejectionReason = reason,
)
}
Expand Down Expand Up @@ -727,12 +724,12 @@ private class JdbcLedgerDao(
completionInfo: Option[state.CompletionInfo],
workflowId: Option[Ref.WorkflowId],
transactionId: Ref.TransactionId,
ledgerEffectiveTime: Instant,
ledgerEffectiveTime: Timestamp,
offsetStep: OffsetStep,
transaction: CommittedTransaction,
divulgedContracts: Iterable[state.DivulgedContract],
blindingInfo: Option[BlindingInfo],
recordTime: Instant,
recordTime: Timestamp,
)(implicit loggingContext: LoggingContext): Future[PersistenceResponse] = {
logger.info("Storing transaction")
dbDispatcher
Expand All @@ -746,7 +743,7 @@ private class JdbcLedgerDao(
state.Update.TransactionAccepted(
optCompletionInfo = completionInfo,
transactionMeta = state.TransactionMeta(
ledgerEffectiveTime = Time.Timestamp.assertFromInstant(ledgerEffectiveTime),
ledgerEffectiveTime = ledgerEffectiveTime,
workflowId = workflowId,
submissionTime = null, // not used for DbDto generation
submissionSeed = null, // not used for DbDto generation
Expand All @@ -756,7 +753,7 @@ private class JdbcLedgerDao(
),
transaction = transaction,
transactionId = transactionId,
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
divulgedContracts = divulgedContracts.toList,
blindingInfo = blindingInfo,
)
Expand All @@ -765,7 +762,7 @@ private class JdbcLedgerDao(
case Some(reason) =>
completionInfo.map(info =>
state.Update.CommandRejected(
recordTime = Time.Timestamp.assertFromInstant(recordTime),
recordTime = recordTime,
completionInfo = info,
reasonTemplate = reason.toStateV2RejectionReason,
)
Expand Down
Loading

0 comments on commit 29d6d39

Please sign in to comment.