Split up Notary protocol into Client and Service parts. The Service protocol can be extended to provide additional transaction processing logic, e.g. validation.

Implemented a Simple and Validating Notary services.

contracts/src/main/kotlin/com/r3corda/protocols/TwoPartyTradeProtocol.kt

@@ -97,7 +97,7 @@ object TwoPartyTradeProtocol {
         @Suspendable
         private fun getNotarySignature(stx: SignedTransaction): DigitalSignature.LegallyIdentifiable {
             progressTracker.currentStep = NOTARY
-            return subProtocol(NotaryProtocol(stx.tx))
+            return subProtocol(NotaryProtocol.Client(stx.tx))
         }
 
         @Suspendable

core/src/main/kotlin/com/r3corda/core/node/services/ServiceType.kt

@@ -1,11 +1,3 @@
-/*
- * Copyright 2016 Distributed Ledger Group LLC.  Distributed as Licensed Company IP to DLG Group Members
- * pursuant to the August 7, 2015 Advisory Services Agreement and subject to the Company IP License terms
- * set forth therein.
- *
- * All other rights reserved.
- */
-
 package com.r3corda.core.node.services
 
 /**
@@ -29,6 +21,8 @@ abstract class ServiceType(val id: String) {
                 false
             }
 
+    fun isSubTypeOf(superType: ServiceType) = (id == superType.id) || id.startsWith(superType.id + ".")
+
     override fun hashCode(): Int = id.hashCode()
     override fun toString(): String = id.toString()
 }

core/src/main/kotlin/com/r3corda/core/node/services/TimestampChecker.kt

@@ -1,4 +1,4 @@
-package com.r3corda.node.services.transactions
+package com.r3corda.core.node.services
 
 import com.r3corda.core.contracts.TimestampCommand
 import com.r3corda.core.seconds
@@ -9,7 +9,7 @@ import java.time.Duration
 /**
  * Checks if the given timestamp falls within the allowed tolerance interval
  */
-class TimestampChecker(val clock: Clock = Clock.systemDefaultZone(),
+class TimestampChecker(val clock: Clock = Clock.systemUTC(),
                        val tolerance: Duration = 30.seconds) {
     fun isValid(timestampCommand: TimestampCommand): Boolean {
         val before = timestampCommand.before

core/src/main/kotlin/com/r3corda/protocols/NotaryProtocol.kt

@@ -1,32 +1,41 @@
 package com.r3corda.protocols
 
 import co.paralleluniverse.fibers.Suspendable
-import com.r3corda.core.crypto.Party
 import com.r3corda.core.contracts.TimestampCommand
 import com.r3corda.core.contracts.WireTransaction
 import com.r3corda.core.crypto.DigitalSignature
+import com.r3corda.core.crypto.Party
 import com.r3corda.core.crypto.SignedData
+import com.r3corda.core.crypto.signWithECDSA
 import com.r3corda.core.messaging.SingleMessageRecipient
 import com.r3corda.core.node.NodeInfo
+import com.r3corda.core.node.services.TimestampChecker
+import com.r3corda.core.node.services.UniquenessException
 import com.r3corda.core.node.services.UniquenessProvider
+import com.r3corda.core.noneOrSingle
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.random63BitValue
 import com.r3corda.core.serialization.SerializedBytes
+import com.r3corda.core.serialization.deserialize
+import com.r3corda.core.serialization.serialize
 import com.r3corda.core.utilities.ProgressTracker
 import com.r3corda.core.utilities.UntrustworthyData
 import java.security.PublicKey
 
-/**
+object NotaryProtocol {
+    val TOPIC = "platform.notary.request"
+    val TOPIC_INITIATE = "platform.notary.initiate"
+
+    /**
      * A protocol to be used for obtaining a signature from a [NotaryService] ascertaining the transaction
      * timestamp is correct and none of its inputs have been used in another completed transaction
      *
      * @throws NotaryException in case the any of the inputs to the transaction have been consumed
      *                         by another transaction or the timestamp is invalid
      */
-class NotaryProtocol(private val wtx: WireTransaction,
+    class Client(private val wtx: WireTransaction,
-                     override val progressTracker: ProgressTracker = NotaryProtocol.tracker()) : ProtocolLogic<DigitalSignature.LegallyIdentifiable>() {
+                 override val progressTracker: ProgressTracker = Client.tracker()) : ProtocolLogic<DigitalSignature.LegallyIdentifiable>() {
         companion object {
-        val TOPIC = "platform.notary.request"
 
             object REQUESTING : ProgressTracker.Step("Requesting signature by Notary service")
 
@@ -42,9 +51,14 @@ class NotaryProtocol(private val wtx: WireTransaction,
             progressTracker.currentStep = REQUESTING
             notaryNode = findNotaryNode()
 
-        val sessionID = random63BitValue()
+            val sendSessionID = random63BitValue()
-        val request = SignRequest(wtx.serialized, serviceHub.storageService.myLegalIdentity, serviceHub.networkService.myAddress, sessionID)
+            val receiveSessionID = random63BitValue()
-        val response = sendAndReceive<Result>(TOPIC, notaryNode.address, 0, sessionID, request)
+
+            val handshake = Handshake(serviceHub.networkService.myAddress, sendSessionID, receiveSessionID)
+            sendAndReceive<Unit>(TOPIC_INITIATE, notaryNode.address, 0, receiveSessionID, handshake)
+
+            val request = SignRequest(wtx.serialized, serviceHub.storageService.myLegalIdentity)
+            val response = sendAndReceive<Result>(TOPIC, notaryNode.address, sendSessionID, receiveSessionID, request)
 
             val notaryResult = validateResponse(response)
             return notaryResult.sig ?: throw NotaryException(notaryResult.error!!)
@@ -84,12 +98,94 @@ class NotaryProtocol(private val wtx: WireTransaction,
             val notaryNode = serviceHub.networkMapCache.getNodeByPublicKey(notaryKey)
             return notaryNode ?: throw IllegalStateException("No Notary node can be found with the specified public key")
         }
+    }
+
+    /**
+     * Checks that the timestamp command is valid (if present) and commits the input state, or returns a conflict
+     * if any of the input states have been previously committed.
+     *
+     * Extend this class, overriding _beforeCommit_ to add custom transaction processing/validation logic.
+     *
+     * TODO: the notary service should only be able to see timestamp commands and inputs
+     */
+    open class Service(val otherSide: SingleMessageRecipient,
+                       val sendSessionID: Long,
+                       val receiveSessionID: Long,
+                       val timestampChecker: TimestampChecker,
+                       val uniquenessProvider: UniquenessProvider) : ProtocolLogic<Unit>() {
+        @Suspendable
+        override fun call() {
+            val request = receive<SignRequest>(TOPIC, receiveSessionID).validate { it }
+            val txBits = request.txBits
+            val reqIdentity = request.callerIdentity
+
+            val wtx = txBits.deserialize()
+            val result: Result
+            try {
+                validateTimestamp(wtx)
+                beforeCommit(wtx, reqIdentity)
+                commitInputStates(wtx, reqIdentity)
+
+                val sig = sign(txBits)
+                result = Result.noError(sig)
+
+            } catch(e: NotaryException) {
+                result = Result.withError(e.error)
+            }
+
+            send(TOPIC, otherSide, sendSessionID, result)
+        }
+
+        private fun validateTimestamp(tx: WireTransaction) {
+            val timestampCmd = try {
+                tx.commands.noneOrSingle { it.value is TimestampCommand } ?: return
+            } catch (e: IllegalArgumentException) {
+                throw NotaryException(NotaryError.MoreThanOneTimestamp())
+            }
+            val myIdentity = serviceHub.storageService.myLegalIdentity
+            if (!timestampCmd.signers.contains(myIdentity.owningKey))
+                throw NotaryException(NotaryError.NotForMe())
+            if (!timestampChecker.isValid(timestampCmd.value as TimestampCommand))
+                throw NotaryException(NotaryError.TimestampInvalid())
+        }
+
+        /**
+         * No pre-commit processing is done. Transaction is not checked for contract-validity, as that would require fully
+         * resolving it into a [TransactionForVerification], for which the caller would have to reveal the whole transaction
+         * history chain.
+         * As a result, the Notary _will commit invalid transactions_ as well, but as it also records the identity of
+         * the caller, it is possible to raise a dispute and verify the validity of the transaction and subsequently
+         * undo the commit of the input states (the exact mechanism still needs to be worked out)
+         */
+        @Suspendable
+        open fun beforeCommit(wtx: WireTransaction, reqIdentity: Party) {
+        }
+
+        private fun commitInputStates(tx: WireTransaction, reqIdentity: Party) {
+            try {
+                uniquenessProvider.commit(tx, reqIdentity)
+            } catch (e: UniquenessException) {
+                val conflictData = e.error.serialize()
+                val signedConflict = SignedData(conflictData, sign(conflictData))
+                throw NotaryException(NotaryError.Conflict(tx, signedConflict))
+            }
+        }
+
+        private fun <T : Any> sign(bits: SerializedBytes<T>): DigitalSignature.LegallyIdentifiable {
+            val mySigningKey = serviceHub.storageService.myLegalIdentityKey
+            val myIdentity = serviceHub.storageService.myLegalIdentity
+            return mySigningKey.signWithECDSA(bits, myIdentity)
+        }
+    }
+
+    class Handshake(
+            replyTo: SingleMessageRecipient,
+            val sendSessionID: Long,
+            sessionID: Long) : AbstractRequestMessage(replyTo, sessionID)
 
     /** TODO: The caller must authenticate instead of just specifying its identity */
     class SignRequest(val txBits: SerializedBytes<WireTransaction>,
-                      val callerIdentity: Party,
+                      val callerIdentity: Party)
-                      replyTo: SingleMessageRecipient,
-                      sessionID: Long) : AbstractRequestMessage(replyTo, sessionID)
 
     data class Result private constructor(val sig: DigitalSignature.LegallyIdentifiable?, val error: NotaryError?) {
         companion object {
@@ -97,6 +193,24 @@ class NotaryProtocol(private val wtx: WireTransaction,
             fun noError(sig: DigitalSignature.LegallyIdentifiable) = Result(sig, null)
         }
     }
+
+    interface Factory {
+        fun create(otherSide: SingleMessageRecipient,
+                   sendSessionID: Long,
+                   receiveSessionID: Long,
+                   timestampChecker: TimestampChecker,
+                   uniquenessProvider: UniquenessProvider): Service
+    }
+
+    object DefaultFactory : Factory {
+        override fun create(otherSide: SingleMessageRecipient,
+                            sendSessionID: Long,
+                            receiveSessionID: Long,
+                            timestampChecker: TimestampChecker,
+                            uniquenessProvider: UniquenessProvider): Service {
+            return Service(otherSide, sendSessionID, receiveSessionID, timestampChecker, uniquenessProvider)
+        }
+    }
 }
 
 class NotaryException(val error: NotaryError) : Exception() {
@@ -115,4 +229,6 @@ sealed class NotaryError {
 
     /** Thrown if the time specified in the timestamp command is outside the allowed tolerance */
     class TimestampInvalid : NotaryError()
+
+    class TransactionInvalid : NotaryError()
 }

core/src/main/kotlin/com/r3corda/protocols/ResolveTransactionsProtocol.kt

@@ -1,7 +1,6 @@
 package com.r3corda.protocols
 
 import co.paralleluniverse.fibers.Suspendable
-import com.r3corda.core.*
 import com.r3corda.core.contracts.*
 import com.r3corda.core.crypto.SecureHash
 import com.r3corda.core.messaging.SingleMessageRecipient

core/src/main/kotlin/com/r3corda/protocols/TwoPartyDealProtocol.kt

@@ -158,7 +158,7 @@ object TwoPartyDealProtocol {
         @Suspendable
         private fun getNotarySignature(stx: SignedTransaction): DigitalSignature.LegallyIdentifiable {
             progressTracker.currentStep = NOTARY
-            return subProtocol(NotaryProtocol(stx.tx))
+            return subProtocol(NotaryProtocol.Client(stx.tx))
         }
 
         open fun signWithOurKey(partialTX: SignedTransaction): DigitalSignature.WithKey {

core/src/main/kotlin/com/r3corda/protocols/ValidatingNotaryProtocol.kt

@@ -0,0 +1,48 @@
+package com.r3corda.protocols
+
+import co.paralleluniverse.fibers.Suspendable
+import com.r3corda.core.contracts.TransactionVerificationException
+import com.r3corda.core.contracts.WireTransaction
+import com.r3corda.core.contracts.toLedgerTransaction
+import com.r3corda.core.crypto.Party
+import com.r3corda.core.messaging.SingleMessageRecipient
+import com.r3corda.core.node.services.TimestampChecker
+import com.r3corda.core.node.services.UniquenessProvider
+import java.security.SignatureException
+
+/**
+ * A notary commit protocol that makes sure a given transaction is valid before committing it. This does mean that the calling
+ * party has to reveal the whole transaction history; however, we avoid complex conflict resolution logic where a party
+ * has its input states "blocked" by a transaction from another party, and needs to establish whether that transaction was
+ * indeed valid
+ */
+class ValidatingNotaryProtocol(otherSide: SingleMessageRecipient,
+                               sessionIdForSend: Long,
+                               sessionIdForReceive: Long,
+                               timestampChecker: TimestampChecker,
+                               uniquenessProvider: UniquenessProvider) : NotaryProtocol.Service(otherSide, sessionIdForSend, sessionIdForReceive, timestampChecker, uniquenessProvider) {
+    @Suspendable
+    override fun beforeCommit(wtx: WireTransaction, reqIdentity: Party) {
+        try {
+            validateDependencies(reqIdentity, wtx)
+            checkContractValid(wtx)
+        } catch (e: Exception) {
+            when (e) {
+                is TransactionVerificationException,
+                is SignatureException -> throw NotaryException(NotaryError.TransactionInvalid())
+                else -> throw e
+            }
+        }
+    }
+
+    private fun checkContractValid(wtx: WireTransaction) {
+        val ltx = wtx.toLedgerTransaction(serviceHub.identityService, serviceHub.storageService.attachments)
+        serviceHub.verifyTransaction(ltx)
+    }
+
+    @Suspendable
+    private fun validateDependencies(reqIdentity: Party, wtx: WireTransaction) {
+        val otherSide = serviceHub.networkMapCache.getNodeByPublicKey(reqIdentity.owningKey)!!.address
+        subProtocol(ResolveTransactionsProtocol(wtx, otherSide))
+    }
+}

node/src/main/kotlin/com/r3corda/node/internal/AbstractNode.kt

@@ -35,7 +35,8 @@ import com.r3corda.node.services.persistence.StorageServiceImpl
 import com.r3corda.node.services.statemachine.StateMachineManager
 import com.r3corda.node.services.transactions.InMemoryUniquenessProvider
 import com.r3corda.node.services.transactions.NotaryService
-import com.r3corda.node.services.transactions.TimestampChecker
+import com.r3corda.node.services.transactions.SimpleNotaryService
+import com.r3corda.node.services.transactions.ValidatingNotaryService
 import com.r3corda.node.services.wallet.NodeWalletService
 import com.r3corda.node.utilities.AddOrRemove
 import com.r3corda.node.utilities.AffinityExecutor
@@ -130,16 +131,14 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         keyManagement = E2ETestKeyManagementService()
         makeInterestRatesOracleService()
         api = APIServerImpl(this)
-
-        // Build services we're advertising
-        if (NetworkMapService.Type in info.advertisedServices) makeNetworkMapService()
-        if (NotaryService.Type in info.advertisedServices) makeNotaryService()
-
         identity = makeIdentityService()
 
         // This object doesn't need to be referenced from this class because it registers handlers on the network
         // service and so that keeps it from being collected.
         DataVendingService(net, storage)
+
+        buildAdvertisedServices()
+
         startMessagingService()
         networkMapRegistrationFuture = registerWithNetworkMap()
         isPreviousCheckpointsPresent = checkpointStorage.checkpoints.any()
@@ -147,6 +146,15 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         started = true
         return this
     }
+
+    private fun buildAdvertisedServices() {
+        val serviceTypes = info.advertisedServices
+        if (NetworkMapService.Type in serviceTypes) makeNetworkMapService()
+
+        val notaryServiceType = serviceTypes.singleOrNull { it.isSubTypeOf(NotaryService.Type) }
+        if (notaryServiceType != null) makeNotaryService(notaryServiceType)
+    }
+
     /**
      * Register this node with the network map cache, and load network map from a remote service (and register for
      * updates) if one has been supplied.
@@ -200,10 +208,15 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         inNodeNetworkMapService = InMemoryNetworkMapService(net, reg, services.networkMapCache)
     }
 
-    open protected fun makeNotaryService() {
+    open protected fun makeNotaryService(type: ServiceType) {
         val uniquenessProvider = InMemoryUniquenessProvider()
         val timestampChecker = TimestampChecker(platformClock, 30.seconds)
-        inNodeNotaryService = NotaryService(net, storage.myLegalIdentity, storage.myLegalIdentityKey, uniquenessProvider, timestampChecker)
+
+        inNodeNotaryService = when (type) {
+            is SimpleNotaryService.Type -> SimpleNotaryService(smm, net, timestampChecker, uniquenessProvider)
+            is ValidatingNotaryService.Type -> ValidatingNotaryService(smm, net, timestampChecker, uniquenessProvider)
+            else -> null
+        }
     }
 
     lateinit var interestRatesService: NodeInterestRates.Service
@@ -246,7 +259,7 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         val checkpointStorage = PerFileCheckpointStorage(dir.resolve("checkpoints"))
         _servicesThatAcceptUploads += attachments
         val (identity, keypair) = obtainKeyPair(dir)
-        return Pair(constructStorageService(attachments, keypair, identity),checkpointStorage)
+        return Pair(constructStorageService(attachments, keypair, identity), checkpointStorage)
     }
 
     protected open fun constructStorageService(attachments: NodeAttachmentService, keypair: KeyPair, identity: Party) =

node/src/main/kotlin/com/r3corda/node/internal/testing/MockNode.kt

@@ -16,7 +16,7 @@ import com.r3corda.node.serialization.NodeClock
 import com.r3corda.node.services.config.NodeConfiguration
 import com.r3corda.node.services.network.InMemoryMessagingNetwork
 import com.r3corda.node.services.network.NetworkMapService
-import com.r3corda.node.services.transactions.NotaryService
+import com.r3corda.node.services.transactions.SimpleNotaryService
 import com.r3corda.node.utilities.AffinityExecutor
 import org.slf4j.Logger
 import java.nio.file.Files
@@ -149,12 +149,12 @@ class MockNetwork(private val threadPerNode: Boolean = false,
     fun createTwoNodes(nodeFactory: Factory = defaultFactory, notaryKeyPair: KeyPair? = null): Pair<MockNode, MockNode> {
         require(nodes.isEmpty())
         return Pair(
-                createNode(null, -1, nodeFactory, true, null, notaryKeyPair, NetworkMapService.Type, NotaryService.Type),
+                createNode(null, -1, nodeFactory, true, null, notaryKeyPair, NetworkMapService.Type, SimpleNotaryService.Type),
                 createNode(nodes[0].info, -1, nodeFactory, true, null)
         )
     }
 
-    fun createNotaryNode(legalName: String? = null, keyPair: KeyPair? = null) = createNode(null, -1, defaultFactory, true, legalName, keyPair, NetworkMapService.Type, NotaryService.Type)
+    fun createNotaryNode(legalName: String? = null, keyPair: KeyPair? = null) = createNode(null, -1, defaultFactory, true, legalName, keyPair, NetworkMapService.Type, SimpleNotaryService.Type)
     fun createPartyNode(networkMapAddr: NodeInfo, legalName: String? = null, keyPair: KeyPair? = null) = createNode(networkMapAddr, -1, defaultFactory, true, legalName, keyPair)
 
     fun addressToNode(address: SingleMessageRecipient): MockNode = nodes.single { it.net.myAddress == address }

node/src/main/kotlin/com/r3corda/node/internal/testing/Simulation.kt

@@ -9,11 +9,11 @@ import com.r3corda.core.node.services.ServiceType
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.then
 import com.r3corda.core.utilities.ProgressTracker
-import com.r3corda.node.services.transactions.NotaryService
 import com.r3corda.node.services.clientapi.NodeInterestRates
 import com.r3corda.node.services.config.NodeConfiguration
 import com.r3corda.node.services.network.InMemoryMessagingNetwork
 import com.r3corda.node.services.network.NetworkMapService
+import com.r3corda.node.services.transactions.SimpleNotaryService
 import rx.Observable
 import rx.subjects.PublishSubject
 import java.nio.file.Path
@@ -82,7 +82,7 @@ abstract class Simulation(val runAsync: Boolean,
     object NotaryNodeFactory : MockNetwork.Factory {
         override fun create(dir: Path, config: NodeConfiguration, network: MockNetwork, networkMapAddr: NodeInfo?,
                             advertisedServices: Set<ServiceType>, id: Int, keyPair: KeyPair?): MockNetwork.MockNode {
-            require(advertisedServices.contains(NotaryService.Type))
+            require(advertisedServices.contains(SimpleNotaryService.Type))
             val cfg = object : NodeConfiguration {
                 override val myLegalName: String = "Notary Service"
                 override val exportJMXto: String = ""
@@ -134,7 +134,7 @@ abstract class Simulation(val runAsync: Boolean,
     val networkMap: SimulatedNode
             = network.createNode(null, nodeFactory = NetworkMapNodeFactory, advertisedServices = NetworkMapService.Type) as SimulatedNode
     val notary: SimulatedNode
-            = network.createNode(networkMap.info, nodeFactory = NotaryNodeFactory, advertisedServices = NotaryService.Type) as SimulatedNode
+            = network.createNode(networkMap.info, nodeFactory = NotaryNodeFactory, advertisedServices = SimpleNotaryService.Type) as SimulatedNode
     val regulators: List<SimulatedNode> = listOf(network.createNode(networkMap.info, start = false, nodeFactory = RegulatorFactory) as SimulatedNode)
     val ratesOracle: SimulatedNode
             = network.createNode(networkMap.info, start = false, nodeFactory = RatesOracleFactory, advertisedServices = NodeInterestRates.Type) as SimulatedNode

node/src/main/kotlin/com/r3corda/node/internal/testing/TestUtils.kt

@@ -1,20 +1,29 @@
-@file:Suppress("UNUSED_PARAMETER")
+package com.r3corda.node.internal.testing
-
-package com.r3corda.node.testutils
 
 import com.r3corda.contracts.DummyContract
 import com.r3corda.core.contracts.StateRef
 import com.r3corda.core.crypto.Party
+import com.r3corda.core.seconds
 import com.r3corda.core.testing.DUMMY_NOTARY
 import com.r3corda.core.testing.DUMMY_NOTARY_KEY
 import com.r3corda.node.internal.AbstractNode
+import java.time.Instant
 import java.util.*
 
 fun issueState(node: AbstractNode, notary: Party = DUMMY_NOTARY): StateRef {
-    val tx = DummyContract().generateInitial(node.info.identity.ref(0), Random().nextInt(), DUMMY_NOTARY)
+    val tx = DummyContract().generateInitial(node.info.identity.ref(0), Random().nextInt(), notary)
     tx.signWith(node.storage.myLegalIdentityKey)
     tx.signWith(DUMMY_NOTARY_KEY)
     val stx = tx.toSignedTransaction()
     node.services.recordTransactions(listOf(stx))
     return StateRef(stx.id, 0)
 }
+
+fun issueInvalidState(node: AbstractNode, notary: Party = DUMMY_NOTARY): StateRef {
+    val tx = DummyContract().generateInitial(node.info.identity.ref(0), Random().nextInt(), notary)
+    tx.setTime(Instant.now(), notary, 30.seconds)
+    tx.signWith(node.storage.myLegalIdentityKey)
+    val stx = tx.toSignedTransaction(false)
+    node.services.recordTransactions(listOf(stx))
+    return StateRef(stx.id, 0)
+}

node/src/main/kotlin/com/r3corda/node/services/network/InMemoryNetworkMapCache.kt

@@ -47,7 +47,7 @@ open class InMemoryNetworkMapCache() : SingletonSerializeAsToken(), NetworkMapCa
     protected var registeredNodes = Collections.synchronizedMap(HashMap<Party, NodeInfo>())
 
     override fun get() = registeredNodes.map { it.value }
-    override fun get(serviceType: ServiceType) = registeredNodes.filterValues { it.advertisedServices.contains(serviceType) }.map { it.value }
+    override fun get(serviceType: ServiceType) = registeredNodes.filterValues { it.advertisedServices.any { it.isSubTypeOf(serviceType) } }.map { it.value }
     override fun getRecommended(type: ServiceType, contract: Contract, vararg party: Party): NodeInfo? = get(type).firstOrNull()
     override fun getNodeByLegalName(name: String) = get().singleOrNull { it.identity.name == name }
     override fun getNodeByPublicKey(publicKey: PublicKey) = get().singleOrNull { it.identity.owningKey == publicKey }

node/src/main/kotlin/com/r3corda/node/services/transactions/NotaryService.kt

@@ -1,100 +1,46 @@
 package com.r3corda.node.services.transactions
 
-import com.r3corda.core.contracts.TimestampCommand
-import com.r3corda.core.contracts.WireTransaction
-import com.r3corda.core.crypto.DigitalSignature
-import com.r3corda.core.crypto.Party
-import com.r3corda.core.crypto.SignedData
-import com.r3corda.core.crypto.signWithECDSA
 import com.r3corda.core.messaging.MessagingService
+import com.r3corda.core.messaging.SingleMessageRecipient
 import com.r3corda.core.node.services.ServiceType
-import com.r3corda.core.node.services.UniquenessException
+import com.r3corda.core.node.services.TimestampChecker
 import com.r3corda.core.node.services.UniquenessProvider
-import com.r3corda.core.noneOrSingle
-import com.r3corda.core.serialization.SerializedBytes
-import com.r3corda.core.serialization.deserialize
-import com.r3corda.core.serialization.serialize
-import com.r3corda.core.utilities.loggerFor
 import com.r3corda.node.services.api.AbstractNodeService
-import com.r3corda.protocols.NotaryError
+import com.r3corda.node.services.statemachine.StateMachineManager
-import com.r3corda.protocols.NotaryException
 import com.r3corda.protocols.NotaryProtocol
-import java.security.KeyPair
 
 /**
  * A Notary service acts as the final signer of a transaction ensuring two things:
  * - The (optional) timestamp of the transaction is valid
  * - None of the referenced input states have previously been consumed by a transaction signed by this Notary
  *
- * A transaction has to be signed by a Notary to be considered valid (except for output-only transactions w/o a timestamp)
+ * A transaction has to be signed by a Notary to be considered valid (except for output-only transactions without a timestamp).
+ *
+ * This is the base implementation that can be customised with specific Notary transaction commit protocol
  */
-class NotaryService(net: MessagingService,
+abstract class NotaryService(val smm: StateMachineManager,
-                    val identity: Party,
+                             net: MessagingService,
-                    val signingKey: KeyPair,
+                             val timestampChecker: TimestampChecker,
-                    val uniquenessProvider: UniquenessProvider,
+                             val uniquenessProvider: UniquenessProvider) : AbstractNodeService(net) {
-                    val timestampChecker: TimestampChecker) : AbstractNodeService(net) {
     object Type : ServiceType("corda.notary")
 
-    private val logger = loggerFor<NotaryService>()
+    abstract val logger: org.slf4j.Logger
+
+    /** Implement a factory that specifies the transaction commit protocol for the notary service to use */
+    abstract val protocolFactory: NotaryProtocol.Factory
 
     init {
-        check(identity.owningKey == signingKey.public)
+        addMessageHandler(NotaryProtocol.TOPIC_INITIATE,
-        addMessageHandler(NotaryProtocol.TOPIC,
+                { req: NotaryProtocol.Handshake -> processRequest(req) }
-                { req: NotaryProtocol.SignRequest -> processRequest(req.txBits, req.callerIdentity) },
-                { message, e -> logger.error("Exception during notary service request processing", e) }
         )
     }
 
-    /**
+    private fun processRequest(req: NotaryProtocol.Handshake) {
-     * Checks that the timestamp command is valid (if present) and commits the input state, or returns a conflict
+        val protocol = protocolFactory.create(req.replyTo as SingleMessageRecipient,
-     * if any of the input states have been previously committed
+                req.sessionID!!,
-     *
+                req.sendSessionID,
-     * Note that the transaction is not checked for contract-validity, as that would require fully resolving it
+                timestampChecker,
-     * into a [TransactionForVerification], for which the caller would have to reveal the whole transaction history chain.
+                uniquenessProvider)
-     * As a result, the Notary _will commit invalid transactions_ as well, but as it also records the identity of
+        smm.add(NotaryProtocol.TOPIC, protocol)
-     * the caller, it is possible to raise a dispute and verify the validity of the transaction and subsequently
-     * undo the commit of the input states (the exact mechanism still needs to be worked out)
-     *
-     * TODO: the notary service should only be able to see timestamp commands and inputs
-     */
-    fun processRequest(txBits: SerializedBytes<WireTransaction>, reqIdentity: Party): NotaryProtocol.Result {
-        val wtx = txBits.deserialize()
-        try {
-            validateTimestamp(wtx)
-            commitInputStates(wtx, reqIdentity)
-        } catch(e: NotaryException) {
-            return NotaryProtocol.Result.withError(e.error)
     }
-
-        val sig = sign(txBits)
-        return NotaryProtocol.Result.noError(sig)
-    }
-
-    private fun validateTimestamp(tx: WireTransaction) {
-        val timestampCmd = try {
-            tx.commands.noneOrSingle { it.value is TimestampCommand } ?: return
-        } catch (e: IllegalArgumentException) {
-            throw NotaryException(NotaryError.MoreThanOneTimestamp())
-        }
-        if (!timestampCmd.signers.contains(identity.owningKey))
-            throw NotaryException(NotaryError.NotForMe())
-        if (!timestampChecker.isValid(timestampCmd.value as TimestampCommand))
-            throw NotaryException(NotaryError.TimestampInvalid())
-    }
-
-    private fun commitInputStates(tx: WireTransaction, reqIdentity: Party) {
-        try {
-            uniquenessProvider.commit(tx, reqIdentity)
-        } catch (e: UniquenessException) {
-            val conflictData = e.error.serialize()
-            val signedConflict = SignedData(conflictData, sign(conflictData))
-            throw NotaryException(NotaryError.Conflict(tx, signedConflict))
-        }
-    }
-
-    private fun <T : Any> sign(bits: SerializedBytes<T>): DigitalSignature.LegallyIdentifiable {
-        return signingKey.signWithECDSA(bits, identity)
-    }
-
 }
-

node/src/main/kotlin/com/r3corda/node/services/transactions/SimpleNotaryService.kt

@@ -0,0 +1,22 @@
+package com.r3corda.node.services.transactions
+
+import com.r3corda.core.messaging.MessagingService
+import com.r3corda.core.node.services.ServiceType
+import com.r3corda.core.node.services.TimestampChecker
+import com.r3corda.core.node.services.UniquenessProvider
+import com.r3corda.core.utilities.loggerFor
+import com.r3corda.node.services.statemachine.StateMachineManager
+import com.r3corda.protocols.NotaryProtocol
+
+/** A simple Notary service that does not perform transaction validation */
+class SimpleNotaryService(
+        smm: StateMachineManager,
+        net: MessagingService,
+        timestampChecker: TimestampChecker,
+        uniquenessProvider: UniquenessProvider) : NotaryService(smm, net, timestampChecker, uniquenessProvider) {
+    object Type : ServiceType("corda.notary.simple")
+
+    override val logger = loggerFor<SimpleNotaryService>()
+
+    override val protocolFactory = NotaryProtocol.DefaultFactory
+}

node/src/main/kotlin/com/r3corda/node/services/transactions/ValidatingNotaryService.kt

@@ -0,0 +1,33 @@
+package com.r3corda.node.services.transactions
+
+import com.r3corda.core.messaging.MessagingService
+import com.r3corda.core.messaging.SingleMessageRecipient
+import com.r3corda.core.node.services.ServiceType
+import com.r3corda.core.node.services.TimestampChecker
+import com.r3corda.core.node.services.UniquenessProvider
+import com.r3corda.core.utilities.loggerFor
+import com.r3corda.node.services.statemachine.StateMachineManager
+import com.r3corda.protocols.NotaryProtocol
+import com.r3corda.protocols.ValidatingNotaryProtocol
+
+/** A Notary service that validates the transaction chain of he submitted transaction before committing it */
+class ValidatingNotaryService(
+        smm: StateMachineManager,
+        net: MessagingService,
+        timestampChecker: TimestampChecker,
+        uniquenessProvider: UniquenessProvider
+) : NotaryService(smm, net, timestampChecker, uniquenessProvider) {
+    object Type : ServiceType("corda.notary.validating")
+
+    override val logger = loggerFor<ValidatingNotaryService>()
+
+    override val protocolFactory = object : NotaryProtocol.Factory {
+        override fun create(otherSide: SingleMessageRecipient,
+                            sendSessionID: Long,
+                            receiveSessionID: Long,
+                            timestampChecker: TimestampChecker,
+                            uniquenessProvider: UniquenessProvider): NotaryProtocol.Service {
+            return ValidatingNotaryProtocol(otherSide, sendSessionID, receiveSessionID, timestampChecker, uniquenessProvider)
+        }
+    }
+}

node/src/test/kotlin/com/r3corda/node/messaging/AttachmentTests.kt

@@ -12,11 +12,11 @@ import com.r3corda.node.internal.testing.MockNetwork
 import com.r3corda.node.services.config.NodeConfiguration
 import com.r3corda.node.services.network.NetworkMapService
 import com.r3corda.node.services.persistence.NodeAttachmentService
-import com.r3corda.node.services.transactions.NotaryService
+import com.r3corda.node.services.transactions.SimpleNotaryService
-import org.junit.Before
-import org.junit.Test
 import com.r3corda.protocols.FetchAttachmentsProtocol
 import com.r3corda.protocols.FetchDataProtocol
+import org.junit.Before
+import org.junit.Test
 import java.io.ByteArrayInputStream
 import java.io.ByteArrayOutputStream
 import java.nio.ByteBuffer
@@ -100,7 +100,7 @@ class AttachmentTests {
                     }
                 }
             }
-        }, true, null, null, NetworkMapService.Type, NotaryService.Type)
+        }, true, null, null, NetworkMapService.Type, SimpleNotaryService.Type)
         val n1 = network.createNode(n0.info)
 
         // Insert an attachment into node zero's store directly.

node/src/test/kotlin/com/r3corda/node/services/NotaryServiceTests.kt

@@ -1,16 +1,19 @@
 package com.r3corda.node.services
 
+import com.r3corda.core.contracts.TimestampCommand
 import com.r3corda.core.contracts.TransactionBuilder
 import com.r3corda.core.seconds
 import com.r3corda.core.testing.DUMMY_NOTARY
 import com.r3corda.core.testing.DUMMY_NOTARY_KEY
 import com.r3corda.node.internal.testing.MockNetwork
-import com.r3corda.node.testutils.issueState
+import com.r3corda.node.internal.testing.issueState
-import org.junit.Before
+import com.r3corda.node.services.network.NetworkMapService
-import org.junit.Test
+import com.r3corda.node.services.transactions.SimpleNotaryService
 import com.r3corda.protocols.NotaryError
 import com.r3corda.protocols.NotaryException
 import com.r3corda.protocols.NotaryProtocol
+import org.junit.Before
+import org.junit.Test
 import java.time.Instant
 import java.util.concurrent.ExecutionException
 import kotlin.test.assertEquals
@@ -22,12 +25,14 @@ class NotaryServiceTests {
     lateinit var notaryNode: MockNetwork.MockNode
     lateinit var clientNode: MockNetwork.MockNode
 
-    @Before
+    @Before fun setup() {
-    fun setup() {
-        // TODO: Move into MockNetwork
         net = MockNetwork()
-        notaryNode = net.createNotaryNode(DUMMY_NOTARY.name, DUMMY_NOTARY_KEY)
+        notaryNode = net.createNode(
-        clientNode = net.createPartyNode(networkMapAddr = notaryNode.info)
+                legalName = DUMMY_NOTARY.name,
+                keyPair = DUMMY_NOTARY_KEY,
+                advertisedServices = *arrayOf(NetworkMapService.Type, SimpleNotaryService.Type)
+        )
+        clientNode = net.createNode(networkMapAddress = notaryNode.info)
         net.runNetwork() // Clear network map registration messages
     }
 
@@ -35,9 +40,9 @@ class NotaryServiceTests {
         val inputState = issueState(clientNode)
         val tx = TransactionBuilder().withItems(inputState)
         tx.setTime(Instant.now(), DUMMY_NOTARY, 30.seconds)
-        var wtx = tx.toWireTransaction()
+        val wtx = tx.toWireTransaction()
 
-        val protocol = NotaryProtocol(wtx, NotaryProtocol.Companion.tracker())
+        val protocol = NotaryProtocol.Client(wtx)
         val future = clientNode.smm.add(NotaryProtocol.TOPIC, protocol)
         net.runNetwork()
 
@@ -49,7 +54,7 @@ class NotaryServiceTests {
         val inputState = issueState(clientNode)
         val wtx = TransactionBuilder().withItems(inputState).toWireTransaction()
 
-        val protocol = NotaryProtocol(wtx, NotaryProtocol.Companion.tracker())
+        val protocol = NotaryProtocol.Client(wtx)
         val future = clientNode.smm.add(NotaryProtocol.TOPIC, protocol)
         net.runNetwork()
 
@@ -61,9 +66,9 @@ class NotaryServiceTests {
         val inputState = issueState(clientNode)
         val tx = TransactionBuilder().withItems(inputState)
         tx.setTime(Instant.now().plusSeconds(3600), DUMMY_NOTARY, 30.seconds)
-        var wtx = tx.toWireTransaction()
+        val wtx = tx.toWireTransaction()
 
-        val protocol = NotaryProtocol(wtx, NotaryProtocol.Companion.tracker())
+        val protocol = NotaryProtocol.Client(wtx)
         val future = clientNode.smm.add(NotaryProtocol.TOPIC, protocol)
         net.runNetwork()
 
@@ -72,14 +77,32 @@ class NotaryServiceTests {
         assertTrue(error is NotaryError.TimestampInvalid)
     }
 
+    @Test fun `should report error for transaction with more than one timestamp`() {
+        val inputState = issueState(clientNode)
+        val tx = TransactionBuilder().withItems(inputState)
+        val timestamp = TimestampCommand(Instant.now(), 30.seconds)
+        tx.addCommand(timestamp, DUMMY_NOTARY.owningKey)
+        tx.addCommand(timestamp, DUMMY_NOTARY.owningKey)
+        val wtx = tx.toWireTransaction()
+
+        val protocol = NotaryProtocol.Client(wtx)
+        val future = clientNode.smm.add(NotaryProtocol.TOPIC, protocol)
+        net.runNetwork()
+
+        val ex = assertFailsWith(ExecutionException::class) { future.get() }
+        val error = (ex.cause as NotaryException).error
+        assertTrue(error is NotaryError.MoreThanOneTimestamp)
+    }
+
     @Test fun `should report conflict for a duplicate transaction`() {
         val inputState = issueState(clientNode)
         val wtx = TransactionBuilder().withItems(inputState).toWireTransaction()
 
-        val firstSpend = NotaryProtocol(wtx)
+        val firstSpend = NotaryProtocol.Client(wtx)
-        val secondSpend = NotaryProtocol(wtx)
+        val secondSpend = NotaryProtocol.Client(wtx)
         clientNode.smm.add("${NotaryProtocol.TOPIC}.first", firstSpend)
         val future = clientNode.smm.add("${NotaryProtocol.TOPIC}.second", secondSpend)
+
         net.runNetwork()
 
         val ex = assertFailsWith(ExecutionException::class) { future.get() }

node/src/test/kotlin/com/r3corda/node/services/TimestampCheckerTests.kt

@@ -1,8 +1,8 @@
 package com.r3corda.node.services
 
 import com.r3corda.core.contracts.TimestampCommand
+import com.r3corda.core.node.services.TimestampChecker
 import com.r3corda.core.seconds
-import com.r3corda.node.services.transactions.TimestampChecker
 import org.junit.Test
 import java.time.Clock
 import java.time.Instant

node/src/test/kotlin/com/r3corda/node/services/ValidatingNotaryServiceTests.kt

@@ -0,0 +1,47 @@
+package com.r3corda.node.services
+
+import com.r3corda.core.contracts.TransactionBuilder
+import com.r3corda.core.testing.DUMMY_NOTARY
+import com.r3corda.core.testing.DUMMY_NOTARY_KEY
+import com.r3corda.node.internal.testing.MockNetwork
+import com.r3corda.node.internal.testing.issueInvalidState
+import com.r3corda.node.services.network.NetworkMapService
+import com.r3corda.node.services.transactions.ValidatingNotaryService
+import com.r3corda.protocols.NotaryError
+import com.r3corda.protocols.NotaryException
+import com.r3corda.protocols.NotaryProtocol
+import org.junit.Before
+import org.junit.Test
+import java.util.concurrent.ExecutionException
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+class ValidatingNotaryServiceTests {
+    lateinit var net: MockNetwork
+    lateinit var notaryNode: MockNetwork.MockNode
+    lateinit var clientNode: MockNetwork.MockNode
+
+    @Before fun setup() {
+        net = MockNetwork()
+        notaryNode = net.createNode(
+                legalName = DUMMY_NOTARY.name,
+                keyPair = DUMMY_NOTARY_KEY,
+                advertisedServices = *arrayOf(NetworkMapService.Type, ValidatingNotaryService.Type)
+        )
+        clientNode = net.createNode(networkMapAddress = notaryNode.info)
+        net.runNetwork() // Clear network map registration messages
+    }
+
+    @Test fun `should report error for invalid transaction dependency`() {
+        val inputState = issueInvalidState(clientNode)
+        val wtx = TransactionBuilder().withItems(inputState).toWireTransaction()
+
+        val protocol = NotaryProtocol.Client(wtx)
+        val future = clientNode.smm.add(NotaryProtocol.TOPIC, protocol)
+        net.runNetwork()
+
+        val ex = assertFailsWith(ExecutionException::class) { future.get() }
+        val notaryError = (ex.cause as NotaryException).error
+        assertTrue(notaryError is NotaryError.TransactionInvalid, "Received wrong Notary error")
+    }
+}

src/main/kotlin/com/r3corda/demos/IRSDemo.kt

@@ -1,24 +1,24 @@
 package com.r3corda.demos
 
 import com.google.common.net.HostAndPort
-import com.typesafe.config.ConfigFactory
 import com.r3corda.core.crypto.Party
 import com.r3corda.core.logElapsedTime
-import com.r3corda.node.internal.Node
-import com.r3corda.node.services.config.NodeConfiguration
-import com.r3corda.node.services.config.NodeConfigurationFromConfig
 import com.r3corda.core.node.NodeInfo
-import com.r3corda.node.services.network.NetworkMapService
-import com.r3corda.node.services.clientapi.NodeInterestRates
-import com.r3corda.node.services.transactions.NotaryService
 import com.r3corda.core.node.services.ServiceType
-import com.r3corda.node.services.messaging.ArtemisMessagingService
 import com.r3corda.core.serialization.deserialize
 import com.r3corda.core.utilities.BriefLogFormatter
 import com.r3corda.demos.api.InterestRateSwapAPI
 import com.r3corda.demos.protocols.AutoOfferProtocol
 import com.r3corda.demos.protocols.ExitServerProtocol
 import com.r3corda.demos.protocols.UpdateBusinessDayProtocol
+import com.r3corda.node.internal.Node
+import com.r3corda.node.services.clientapi.NodeInterestRates
+import com.r3corda.node.services.config.NodeConfiguration
+import com.r3corda.node.services.config.NodeConfigurationFromConfig
+import com.r3corda.node.services.messaging.ArtemisMessagingService
+import com.r3corda.node.services.network.NetworkMapService
+import com.r3corda.node.services.transactions.SimpleNotaryService
+import com.typesafe.config.ConfigFactory
 import joptsimple.OptionParser
 import java.nio.file.Files
 import java.nio.file.Path
@@ -65,12 +65,12 @@ fun main(args: Array<String>) {
 
     val networkMapId = if (options.valueOf(networkMapNetAddr).equals(options.valueOf(networkAddressArg))) {
         // This node provides network map and notary services
-        advertisedServices = setOf(NetworkMapService.Type, NotaryService.Type)
+        advertisedServices = setOf(NetworkMapService.Type, SimpleNotaryService.Type)
         null
     } else {
         advertisedServices = setOf(NodeInterestRates.Type)
         try {
-            nodeInfo(options.valueOf(networkMapNetAddr), options.valueOf(networkMapIdentityFile), setOf(NetworkMapService.Type, NotaryService.Type))
+            nodeInfo(options.valueOf(networkMapNetAddr), options.valueOf(networkMapIdentityFile), setOf(NetworkMapService.Type, SimpleNotaryService.Type))
         } catch (e: Exception) {
             null
         }

src/main/kotlin/com/r3corda/demos/TraderDemo.kt

@@ -24,7 +24,7 @@ import com.r3corda.node.services.config.NodeConfigurationFromConfig
 import com.r3corda.node.services.messaging.ArtemisMessagingService
 import com.r3corda.node.services.network.NetworkMapService
 import com.r3corda.node.services.persistence.NodeAttachmentService
-import com.r3corda.node.services.transactions.NotaryService
+import com.r3corda.node.services.transactions.SimpleNotaryService
 import com.r3corda.node.services.wallet.NodeWalletService
 import com.r3corda.node.utilities.ANSIProgressRenderer
 import com.r3corda.protocols.NotaryProtocol
@@ -114,7 +114,7 @@ fun main(args: Array<String>) {
     // the map is not very helpful, but we need one anyway. So just make the buyer side run the network map as it's
     // the side that sticks around waiting for the seller.
     val networkMapId = if (role == Role.BUYER) {
-        advertisedServices = setOf(NetworkMapService.Type, NotaryService.Type)
+        advertisedServices = setOf(NetworkMapService.Type, SimpleNotaryService.Type)
         null
     } else {
         // In a real system, the identity file of the network map would be shipped with the server software, and there'd
@@ -350,7 +350,7 @@ class TraderDemoProtocolSeller(val myAddress: HostAndPort,
             tx.signWith(keyPair)
 
             // Get the notary to sign it, thus committing the outputs.
-            val notarySig = subProtocol(NotaryProtocol(tx.toWireTransaction()))
+            val notarySig = subProtocol(NotaryProtocol.Client(tx.toWireTransaction()))
             tx.addSignatureUnchecked(notarySig)
 
             // Commit it to local storage.
@@ -365,7 +365,7 @@ class TraderDemoProtocolSeller(val myAddress: HostAndPort,
             val builder = TransactionBuilder()
             CommercialPaper().generateMove(builder, issuance.tx.outRef(0), ownedBy)
             builder.signWith(keyPair)
-            builder.addSignatureUnchecked(subProtocol(NotaryProtocol(builder.toWireTransaction())))
+            builder.addSignatureUnchecked(subProtocol(NotaryProtocol.Client(builder.toWireTransaction())))
             val tx = builder.toSignedTransaction(true)
             serviceHub.recordTransactions(listOf(tx))
             tx