Overview of the Corda State Machine

The state machine has an event feed, and one or more execution threads. Events trigger transitions between the possible states.

Every flow starts in the Pending state as a start flow message, and then transitions to running. In the easiest case, it finally reaches the end of its run function and transitions to success, the result gets returned and any checkpoint data cleaned up.

It can at any point be suspended (more on that below), and then revived later to carry on running.

If it errors, it moves to the FlowHospital, where three things can happen:

• • If the error is a transient error, it will go back to pending to be retried from the last checkpoint
  • If the error is fatal, the flow terminates as failed, an error is returned to the caller and all checkpoints are deleted

• If the error can’t be simply retried, but failing is not an option (e.g. because that would threaten the integrity of the ledger), the flow moves into a hold state. This requires human intervention to
  • fix the condition that led to the error
  • restart the flow from the last checkpoint manually.

Lifecycle of a flow

When thinking about the lifecycle of a flow, it is useful to think about user and kernel space inside a Corda node, where user space is the code run in the flow logic that users can provide, and kernel space is the state machine/flow framework logic around that that facilitates the execution of the run.

1 Start a flow

When a flow gets started via RPC,StartFlow gets called on the state machine. This schedules an event to start the flow (note that RPC handling and message deserialisation happen on a different thread pool). As soon as an execution thread becomes available, the state machine will process the message and transition from Pending to Running. This includes:

• Acquire the thread
• Acquire a database connection and open a transaction
• Instantiate the requested FlowLogic instance from the user CorDapp, including passing in any parameters
• Call the run function on the FlowLogic.

2 Run and suspend

Then the user code runs in the flow until it hits a point where a suspension is required. The most common reasons for suspension are:

• Sending and then waiting for a message
• Calling FlowLogic.sleep()– i.e. suspending the flow for a given amount of time
• Using the flow async API

For this example, let’s assume the flow needs to communicate with a counterparty and calls sendAndReceive(). This function calls out to kernel space and does the following things:

• Serialising the payload passed as a message to this function
• Putting this message onto the send queue where it will get picked up by the P2P messaging system. That system requires its own discussion page, so we’ll only mention here that we require this to be durable, deliver exactly once: Once we posted the message, we assume it will be delivered eventually, and we will in due course get a reply, which will not be duplicated.
• Then it calls Quasar’s parkAndSerialize function, triggering the behaviour described above, recording the stack, values, objects and thread local storage.
• On top of that, the node also needs to deal with allocated resources that Quasar ignores. Most prominently, it needs to commit the open database transaction and release the database connection. Any other resources (e.g. locks) need to be dealt with as well. This is done in Corda code.
• Corda also stores a copy of the fiber checkpoint in the database so that a suspended flow can survive a restart of the node.
  • This uses a fully reflective Kryo serialiser so anything found on the stack can be serialised. This is a different serializer from the ones used in RPC and P2P communication, or for persisting states to the database – those need to limit what they will accept for serialisation and deserialisation, whereas the checkpointing serialiser needs to deal with everything we throw at it and does not need to deal with the long term evolution of data formats and structures.
  • There is a list of things (mostly node/state machine infrastructure) that needs to be ignored for serialisation and will just be there at deserialisation time.
  • Certain classes (e.g. Corda Services) implement the SerializeAsToken interface which means a token is recorded instead of the actual class instance. These classes need to provide a mechanism to get hold of an instance at deserialisation time. This is particularly useful for singletons of which we cannot have multiple copies.
• Then the fiber transitions from running to suspended, and the execution thread becomes available to run other tasks.

3 Receive and resume

Once the response message arrives, the P2P messaging system will put a message in the event queue such that the fiber can be woken up. When an execution thread becomes available, the state transition for the suspended fiber begins.

• The call stack and variables are recreated
• Resources are reacquired, e.g. DB connection and a transaction is opened
• The message is deserialized
• The message will only be acknowledged in Artemis when the flow hits the next suspension point or finishes. As acknowledging will remove it from the durable P2P messaging, this means that in case of failover, the flow will rerun from the checkpoint from after sending the message and re-receive the message from the durable queue.
• Then the flow logic in user space gets invoked at the point where it called sendAndReceive and get the content of the message as return value of the function call.

4 Run to completion

When the flow logic successfully reaches the end of its run function, control passes back to the kernel space.

• The database transaction is committed
• All checkpoints for this flow are removed from the database (as it succeeded, a partial rerun will not be required).
• Any outstanding messages need to acknowledged in Artemis so they’re cleared from the queue.
• The return value is handed over to the RPC pool so it can be serialized and sent back to the calling client.
• The database connection is released, the fiber is terminated and the thread is freed up to run other tasks.

Implications

This above implementation has some implications for developers of CorDapps

• Normal ACID database logic cannot be simply used in CorDapps, even though we expose a raw JDBC connection.
  • You cannot simply commit a transaction in a flow logic run method (calling FlowLogic.sleep(1.millis)is a possible work-around)
  • A suspension can cause a database commit at a surprising time.
  • If the flow fails, the transaction will be rolled back – if you did something with the JDBC connection, that will be lost. On retry, the flow will restart from the last message, so the database transaction will re-occur.
  • You cannot roll back the transaction – inside the flow logic you are inside the flow’s transaction, so if you write SQL code and something goes wrong and you roll back, you could threaten the integrity of the ledger
• Every part of the flow gets run at least once – however, if something goes wrong, the part between the last checkpoint and the point where the error happened will be rerun. The flow clears up any database transaction happening via a rollback so it can be re-run cleanly, but any other interaction with third-party components has to be idempotent or able to handle being called more than once.
• Annotation with @Suspendable has to be done very diligently. Due to the logic Quasar uses, it is not possible to call a suspendable function from a non-suspendable function, as this might lead to a non-supendable function on the callstack during a suspend operation. (The other way round is fine – a suspendable function can call un-interuptible things like e.g. database operations).
• The byte code scanning/rewriting takes quite a bit of time at start-up, the suspend operation is not exactly cheap and so on.
• It is possible to write code where the instrumentation goes wrong (e.g. using suspendable methods inside a sufficiently complex Kotlin .map{} operation). This can result in invalid byte code and the JVM just giving up at runtime.
  This is a known issue with Kotlin (https://youtrack.jetbrains.com/issue/KT-19251) and there is a now a compiler flag you can set to make it work more smoothly once you are aware of the issue:

compileKotlin {
     kotlinOptions { 
          freeCompilerArgs = ["-Xnormalize-constructor-calls=enable"]
     }
}