CORDA-1840 Smarter checkpoint writing (#3677)

* Explicitly add/update checkpoints rather than calling `addOrUpdate` which will run a `select` statement and then a add or update statement.

* Use `currentState.isAnyCheckpointPersisted()` to check for previous checkpoints

node/src/main/kotlin/net/corda/node/services/api/CheckpointStorage.kt

@@ -11,10 +11,16 @@ import java.util.stream.Stream
  */
 interface CheckpointStorage {
     /**
-     * Add a new checkpoint to the store.
+     * Add a checkpoint for a new id to the store. Will throw if there is already a checkpoint for this id
      */
     fun addCheckpoint(id: StateMachineRunId, checkpoint: SerializedBytes<Checkpoint>)
 
+    /**
+     * Update an existing checkpoint. Will throw if there is not checkpoint for this id.
+     */
+    fun updateCheckpoint(id: StateMachineRunId, checkpoint: SerializedBytes<Checkpoint>)
+
+
     /**
      * Remove existing checkpoint from the store.
      * @return whether the id matched a checkpoint that was removed.

node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt

@@ -36,13 +36,22 @@ class DBCheckpointStorage : CheckpointStorage {
     )
 
     override fun addCheckpoint(id: StateMachineRunId, checkpoint: SerializedBytes<Checkpoint>) {
-        currentDBSession().saveOrUpdate(DBCheckpoint().apply {
+        currentDBSession().save(DBCheckpoint().apply {
             checkpointId = id.uuid.toString()
             this.checkpoint = checkpoint.bytes
             log.debug { "Checkpoint $checkpointId, size=${this.checkpoint.size}" }
         })
     }
 
+    override fun updateCheckpoint(id: StateMachineRunId, checkpoint: SerializedBytes<Checkpoint>) {
+        currentDBSession().update(DBCheckpoint().apply {
+            checkpointId = id.uuid.toString()
+            this.checkpoint = checkpoint.bytes
+            log.debug { "Checkpoint $checkpointId, size=${this.checkpoint.size}" }
+        })
+    }
+
+
     override fun removeCheckpoint(id: StateMachineRunId): Boolean {
         val session = currentDBSession()
         val criteriaBuilder = session.criteriaBuilder

node/src/main/kotlin/net/corda/node/services/statemachine/Action.kt

@@ -39,7 +39,7 @@ sealed class Action {
     /**
      * Persist the specified [checkpoint].
      */
-    data class PersistCheckpoint(val id: StateMachineRunId, val checkpoint: Checkpoint) : Action()
+    data class PersistCheckpoint(val id: StateMachineRunId, val checkpoint: Checkpoint, val isCheckpointUpdate: Boolean) : Action()
 
     /**
      * Remove the checkpoint corresponding to [id].

node/src/main/kotlin/net/corda/node/services/statemachine/ActionExecutorImpl.kt

@@ -97,7 +97,11 @@ class ActionExecutorImpl(
     @Suspendable
     private fun executePersistCheckpoint(action: Action.PersistCheckpoint) {
         val checkpointBytes = serializeCheckpoint(action.checkpoint)
-        checkpointStorage.addCheckpoint(action.id, checkpointBytes)
+        if (action.isCheckpointUpdate) {
+            checkpointStorage.updateCheckpoint(action.id, checkpointBytes)
+        } else {
+            checkpointStorage.addCheckpoint(action.id, checkpointBytes)
+        }
         checkpointingMeter.mark()
         checkpointSizesThisSecond.update(checkpointBytes.size.toLong())
         var lastUpdateTime = lastBandwidthUpdate.get()

node/src/main/kotlin/net/corda/node/services/statemachine/transitions/TopLevelTransition.kt

@@ -162,7 +162,7 @@ class TopLevelTransition(
                 )
             } else {
                 actions.addAll(arrayOf(
-                        Action.PersistCheckpoint(context.id, newCheckpoint),
+                        Action.PersistCheckpoint(context.id, newCheckpoint, isCheckpointUpdate = currentState.isAnyCheckpointPersisted),
                         Action.PersistDeduplicationFacts(currentState.pendingDeduplicationHandlers),
                         Action.CommitTransaction,
                         Action.AcknowledgeMessages(currentState.pendingDeduplicationHandlers),

node/src/main/kotlin/net/corda/node/services/statemachine/transitions/UnstartedFlowTransition.kt

@@ -78,7 +78,7 @@ class UnstartedFlowTransition(
     private fun TransitionBuilder.createInitialCheckpoint() {
         actions.addAll(arrayOf(
                 Action.CreateTransaction,
-                Action.PersistCheckpoint(context.id, currentState.checkpoint),
+                Action.PersistCheckpoint(context.id, currentState.checkpoint, isCheckpointUpdate = currentState.isAnyCheckpointPersisted),
                 Action.PersistDeduplicationFacts(currentState.pendingDeduplicationHandlers),
                 Action.CommitTransaction,
                 Action.AcknowledgeMessages(currentState.pendingDeduplicationHandlers)