trait ProcessingSteps[SubmissionParam, SubmissionResult, RequestViewType <: ViewType, Result <: SignedProtocolMessageContent, SubmissionError <: WrapsProcessorError] extends AnyRef
Interface for processing steps that are specific to request types. The ProtocolProcessor wires up these steps with the necessary synchronization and state management, including common processing steps.
Every phase has one main entry method (Phase i, step 1), which produces data for the ProtocolProcessor, The phases also have methods to be called using the results from previous methods for each step.
- SubmissionParam
The bundled submission parameters
- SubmissionResult
The bundled submission results
- RequestViewType
The type of view trees used by the request
- Result
The specific type of the result message
- SubmissionError
The type of errors that can occur during submission processing
- Alphabetic
- By Inheritance
- ProcessingSteps
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Type Members
- case class CheckActivenessAndWritePendingContracts(activenessSet: ActivenessSet, pendingContracts: Seq[WithTransactionId[SerializableContract]], pendingDataAndResponseArgs: PendingDataAndResponseArgs) extends Product with Serializable
Phase 3
Phase 3
- activenessSet
The activeness set for the activeness check
- pendingContracts
The pending contracts to be added to the com.digitalasset.canton.participant.store.StoredContractManager, along with the com.digitalasset.canton.protocol.TransactionId that created the contract
- pendingDataAndResponseArgs
The implementation-specific arguments needed to create the pending data and response
- case class CommitAndStoreContractsAndPublishEvent(commitSet: Option[Future[CommitSet]], contractsToBeStored: Set[LfContractId], maybeEvent: Option[TimestampedEvent], update: Option[CausalityUpdate]) extends Product with Serializable
Phase 7, step 3:
Phase 7, step 3:
- commitSet
scala.None$ if the request should be rejected scala.Some$ a future that will produce the commit set for updating the active contract store
- contractsToBeStored
The contracts to be persisted to the contract store. Must be a subset of the contracts produced in Phase 3, step 2 in CheckActivenessAndWritePendingContracts.
- maybeEvent
The event to be published via the com.digitalasset.canton.participant.event.RecordOrderPublisher
- type DecryptedView = RequestViewType.View
The type of decrypted view trees
- case class DecryptedViews(views: Seq[WithRecipients[DecryptedView]], decryptionErrors: Seq[EncryptedViewMessageDecryptionError[RequestViewType]]) extends Product with Serializable
Phase 3:
Phase 3:
- views
The successfully decrypted views
- decryptionErrors
The decryption errors while trying to decrypt the views
- abstract type PendingDataAndResponseArgs
The type of data needed to generate the pending data and response in constructPendingDataAndResponse.
The type of data needed to generate the pending data and response in constructPendingDataAndResponse. The data is created by decryptViews
- abstract type PendingRequestData <: ProcessingSteps.PendingRequestData
The type of data stored after request processing to make it available for result processing
- abstract type PendingSubmissionData
The data stored for submissions that have been sent out, if any
- abstract type PendingSubmissionId
The type used for look-ups into the PendingSubmissions
- abstract type PendingSubmissions
A store of data on submissions that have been sent out, if any
- trait PreparedBatch extends AnyRef
The actual batch to be sent for a TrackedSubmission
- abstract type RejectionArgs
The type of data needed to create a rejection event in createRejectionEvent.
The type of data needed to create a rejection event in createRejectionEvent. Created by constructPendingDataAndResponse
- type RequestBatch = RequestAndRootHashMessage[OpenEnvelope[EncryptedViewMessage[RequestViewType]]]
The type of request messages
- abstract type RequestError <: WrapsProcessorError
The type of errors that can occur during request processing
- abstract type ResultError <: WrapsProcessorError
The type of errors that can occur during result processing
- case class StorePendingDataAndSendResponseAndCreateTimeout(pendingData: PendingRequestData, mediatorResponses: Seq[(MediatorResponse, Recipients)], causalityMessages: Seq[(CausalityMessage, Recipients)], rejectionArgs: RejectionArgs) extends Product with Serializable
Phase 3:
Phase 3:
- pendingData
The PendingRequestData to be stored until Phase 7
- mediatorResponses
The responses to be sent to the mediator
- rejectionArgs
The implementation-specific arguments needed to create a rejection event on timeout
- sealed trait Submission extends AnyRef
- abstract type SubmissionResultArgs
The type of data needed to generate the submission result in createSubmissionResult.
The type of data needed to generate the submission result in createSubmissionResult. The data is created by updatePendingSubmissions.
- abstract type SubmissionSendError
The submission errors that can occur during sending the batch to the sequencer and updating the pending submission map.
- trait TrackedSubmission extends Submission
Submission to be tracked in-flight and with deduplication.
Submission to be tracked in-flight and with deduplication.
The actual batch to be sent is computed only later by TrackedSubmission.prepareBatch so that tracking information (e.g., the chosen deduplication period) can be incorporated into the batch.
- trait UntrackedSubmission extends Submission
Submission to be sent off without tracking the in-flight submission and without deduplication.
Abstract Value Members
- abstract def computeActivenessSetAndPendingContracts(ts: CantonTimestamp, rc: RequestCounter, sc: SequencerCounter, decryptedViews: NonEmptyList[WithRecipients[DecryptedView]], malformedPayloads: Seq[MalformedPayload], snapshot: DomainSnapshotSyncCryptoApi)(implicit traceContext: TraceContext): EitherT[Future, RequestError, CheckActivenessAndWritePendingContracts]
Phase 3, step 2 (some good views):
Phase 3, step 2 (some good views):
- ts
The timestamp of the request
- rc
The com.digitalasset.canton.participant.RequestCounter of the request
- sc
The com.digitalasset.canton.SequencerCounter of the request
- decryptedViews
The decrypted views from step 1 with the right root hash
- malformedPayloads
The decryption errors and decrypted views with a wrong root hash
- snapshot
Snapshot of the topology state at the request timestamp
- returns
The activeness set and the pending contracts to add to the com.digitalasset.canton.participant.store.StoredContractManager, and the arguments for step 2.
- abstract def constructPendingDataAndResponse(pendingDataAndResponseArgs: PendingDataAndResponseArgs, transferLookup: TransferLookup, contractLookup: ContractLookup, causalityLookup: SingleDomainCausalTracker, activenessResultFuture: Future[ActivenessResult], pendingCursor: Future[Unit], mediatorId: MediatorId)(implicit traceContext: TraceContext): EitherT[Future, RequestError, StorePendingDataAndSendResponseAndCreateTimeout]
Phase 3, step 3:
Phase 3, step 3:
- pendingDataAndResponseArgs
Implementation-specific data passed from decryptViews
- transferLookup
Read-only interface of the com.digitalasset.canton.participant.store.memory.TransferCache
- contractLookup
Read-only interface to the com.digitalasset.canton.participant.store.StoredContractManager, which includes the pending contracts
- activenessResultFuture
Future of the result of the activeness check<
- pendingCursor
Future to complete when the com.digitalasset.canton.participant.protocol.RequestJournal's cursor for the com.digitalasset.canton.participant.protocol.RequestJournal.RequestState.Pending reaching the current request. Block on this future to ensure that all earlier contract store writes are visible.
- mediatorId
The mediator that handles this request
- returns
Returns the PendingRequestData to be stored until Phase 7 and the responses to be sent to the mediator.
- abstract def createRejectionEvent(rejectionArgs: RejectionArgs)(implicit traceContext: TraceContext): Either[ResultError, Option[TimestampedEvent]]
Phase 3, step 4:
Phase 3, step 4:
- rejectionArgs
The implementation-specific information needed for the creation of the rejection event
- abstract def createSubmissionResult(deliver: Deliver[Envelope[_]], submissionResultArgs: SubmissionResultArgs): SubmissionResult
Phase 1, step 3:
- abstract def decryptViews(batch: NonEmptyList[OpenEnvelope[EncryptedViewMessage[RequestViewType]]], snapshot: DomainSnapshotSyncCryptoApi)(implicit traceContext: TraceContext): EitherT[Future, RequestError, DecryptedViews]
Phase 3, step 1:
Phase 3, step 1:
- batch
The batch of messages in the request to be processed
- snapshot
Snapshot of the topology state at the request timestamp
- returns
The decrypted views and the errors encountered during decryption
- abstract def embedNoMediatorError(error: NoMediatorError): SubmissionError
Convert com.digitalasset.canton.participant.protocol.ProtocolProcessor.NoMediatorError into a submission error
- abstract def embedRequestError(err: RequestProcessingError): RequestError
Wrap an error in request processing from the generic request processor
- abstract def embedResultError(err: ResultProcessingError): ResultError
Wrap an error in result processing from the generic request processor
- abstract def eventAndSubmissionIdForInactiveMediator(ts: CantonTimestamp, rc: RequestCounter, sc: SequencerCounter, decryptedViews: NonEmptyList[WithRecipients[DecryptedView]])(implicit traceContext: TraceContext): (Option[TimestampedEvent], Option[PendingSubmissionId])
Phase 3, step 2 (some good views, but the chosen mediator is inactive)
Phase 3, step 2 (some good views, but the chosen mediator is inactive)
- ts
The timestamp of the request
- rc
The com.digitalasset.canton.participant.RequestCounter of the request
- sc
The com.digitalasset.canton.SequencerCounter of the request
- decryptedViews
The decrypted views from step 1 with the right root hash
- returns
The optional rejection event to be published in the event log, and the optional submission ID corresponding to this request
- abstract def getCommitSetAndContractsToBeStoredAndEvent(event: SignedContent[Deliver[DefaultOpenEnvelope]], result: Either[MalformedMediatorRequestResult, Result], pendingRequestData: PendingRequestData, pendingSubmissions: PendingSubmissions, tracker: SingleDomainCausalTracker, hashOps: HashOps)(implicit traceContext: TraceContext): EitherT[Future, ResultError, CommitAndStoreContractsAndPublishEvent]
Phase 7, step 2:
Phase 7, step 2:
- event
The signed com.digitalasset.canton.sequencing.protocol.Deliver event containing the mediator result. It is ensured that the
event
contains exactly one com.digitalasset.canton.protocol.messages.MediatorResult- result
The unpacked mediator result that is contained in the
event
- pendingRequestData
The PendingRequestData produced in Phase 3
- pendingSubmissions
The data stored on submissions in the PendingSubmissions
- returns
The com.digitalasset.canton.participant.protocol.conflictdetection.CommitSet, the contracts from Phase 3 to be persisted to the contract store, and the event to be published
- abstract def pendingDataAndResponseArgsForMalformedPayloads(ts: CantonTimestamp, rc: RequestCounter, sc: SequencerCounter, malformedPayloads: Seq[MalformedPayload], snapshot: DomainSnapshotSyncCryptoApi): Either[RequestError, PendingDataAndResponseArgs]
Phase 3, step 2 (only malformed payloads):
Phase 3, step 2 (only malformed payloads):
Called when no decrypted view has the right root hash.
- returns
scala.Left$ aborts processing and scala.Right$ continues processing with an empty activeness set and no pending contracts
- abstract def pendingRequestMap: (SyncDomainEphemeralState) => Map[RequestId, PendingRequestDataOrReplayData[PendingRequestData]]
Selector for the storage slot for PendingRequestData
- abstract def pendingSubmissions(state: SyncDomainEphemeralState): PendingSubmissions
Selector to get the PendingSubmissions, if any
- abstract def postProcessResult(verdict: Verdict, pendingSubmission: PendingSubmissionData)(implicit traceContext: TraceContext): Unit
Phase 7, step 4:
Phase 7, step 4:
Called after the request reached the state com.digitalasset.canton.participant.protocol.RequestJournal.RequestState.Clean in the request journal, if the participant is the submitter. Also called if a timeout occurs with com.digitalasset.canton.protocol.messages.Verdict.Timeout.
- verdict
The verdict on the request
- abstract def postProcessSubmissionForInactiveMediator(declaredMediator: MediatorId, timestamp: CantonTimestamp, pendingSubmission: PendingSubmissionData)(implicit traceContext: TraceContext): Unit
Phase 3, step 2 (submission where the chosen mediator is inactive)
Phase 3, step 2 (submission where the chosen mediator is inactive)
Called if the chosen mediator is inactive and eventAndSubmissionIdForInactiveMediator returned a submission ID that was pending.
- pendingSubmission
The PendingSubmissionData for the submission ID returned by eventAndSubmissionIdForInactiveMediator
- See also
com.digitalasset.canton.participant.protocol.ProcessingSteps.postProcessResult
- abstract def prepareSubmission(param: SubmissionParam, mediatorId: MediatorId, ephemeralState: SyncDomainEphemeralStateLookup, recentSnapshot: DomainSnapshotSyncCryptoApi)(implicit traceContext: TraceContext): EitherT[FutureUnlessShutdown, SubmissionError, Submission]
Phase 1, step 1:
Phase 1, step 1:
- param
The parameter object encapsulating the parameters of the submit method
- mediatorId
The mediator ID to use for this submission
- ephemeralState
Read-only access to the com.digitalasset.canton.participant.store.SyncDomainEphemeralState
- recentSnapshot
A recent snapshot of the topology state to be used for submission
- abstract def removePendingSubmission(pendingSubmissions: PendingSubmissions, pendingSubmissionId: PendingSubmissionId): Option[PendingSubmissionData]
Phase 1, step 4; and Phase 7, step 1:
Phase 1, step 4; and Phase 7, step 1:
Remove the pending submission from the pending submissions. Called when sending the submission failed or did not lead to a result in time or a result has arrived for the request.
- abstract def requestKind: String
The kind of request, used for logging and error reporting
- abstract def submissionDescription(param: SubmissionParam): String
Extract a description for a submission, used for logging and error reporting
- abstract def submissionIdOfPendingRequest(pendingData: PendingRequestData): PendingSubmissionId
Extract the submission ID that corresponds to a pending request, if any
- abstract def updatePendingSubmissions(pendingSubmissions: PendingSubmissions, submissionParam: SubmissionParam, pendingSubmissionId: PendingSubmissionId): EitherT[Future, SubmissionSendError, SubmissionResultArgs]
Phase 1, step 2:
Phase 1, step 2:
- pendingSubmissions
Stateful store to be updated with data on the pending submission
- submissionParam
Implementation-specific details on the submission, used for logging
- pendingSubmissionId
The key used for updates to the pendingSubmissions
Concrete Value Members
- final def !=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def ##: Int
- Definition Classes
- AnyRef → Any
- final def ==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def asInstanceOf[T0]: T0
- Definition Classes
- Any
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @native() @HotSpotIntrinsicCandidate()
- final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def toString(): String
- Definition Classes
- AnyRef → Any
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- object DecryptedViews extends Serializable