XDB Proposal & Design
This docs will describe the overall design of the xdb project. It will starts with the goal of the project, and then the technologies that it will be building on top of as tech stack. Then it will go to the high level design of how all the components work together. Finally it comes with a background section, and a FAQ section.
XDB is not a new database. It's "eXtending a database with an async programming model".
This means nothing will be dropped from the database that it extended from -- all the existing functionalities of a database will be kept. With this extension, a database will provide both synchronous and asynchronous programming experience.
Internally, xdb utilize database CDC(change data capture) + message queue as implementation, which is a popular design pattern for async programming in the industry. Therefore, from another point of view, xdb creates an abstraction of this design pattern, so that users don't need to implement this pattern themselves(which requires to deal with a lot of low level details of a database).
As of today, there are many excellent database products -- from SQL to NoSQL, and to even "NewSQL". They all provide "synchronous" APIs for applications. For example, a request is made from application to insert a record/row/document, then a database will process and respond back the result synchronously.
This is great but far not enough. The real-world applications often have to deal with "asynchronous process". For example, placing an order means inserting some database records first, then call external service for sending emails, reminders, or tracking order status. Those operations after inserting the database records are "asynchronous":
- There is no builtin/easy way to atomically inserting database records and also call external service
- Communicating with other services often need backoff retry for resiliency (eventual consistency)
- It may need to wait for some duration (like sending reminder often need to use a "durable timer")
- The operations need to be orchestrated with some sequence based on business need
Additionally, those async operations are usually coupling with databases, because databases are usually the "source of truth" for read/write. For example, user may want to update a record if user click the link in a reminder email, or update the record when the order status moved from "shipped" to "delivered".
Today to build an reliable and scalable async process:
- People put queues to represent different steps, and using consumer to executing the steps asynchronously
- People build their own "durable timer services" to persist a timer, or use cron job to run periodically to hack/work around
- People use "workflow" services or programming model like "Step Functions/Conductor/iWF" to orchestrate the API for databases/services
- People use database CDC(change data capture) + message queue to build async logic after database changes
Those solutions are complicated:
- Business logic are scattered across different places, hard to maintain
- Have to deal with a lot of low level APIs(like the CDC message schema)
- May require a lot of work to sync/transport data between different services and database
What if a database can provide async programming model out of the box?
- Users will not need different dependencies to build applications
- No need to sync/transport data between databases and others
The iWF programming model is a perfect fit for this idea. It could work seamlessly with any database.
In iWF framework, users will define a "persistence schema" for storing custom data as "data attributes". Then other two basic components of an iWF workflow: "WorkflowState" and "RPC", user will write code to read and write the data. In the python SDK example(because python is shorter. Java and Golang SDKs are also available in iWF):
class SubmitState(WorkflowState[Form]):
def execute(self, ctx: WorkflowContext, input: Form, command_results: CommandResults, persistence: Persistence,
communication: Communication,
) -> StateDecision:
persistence.set_data_attribute(data_attribute_form, input)
persistence.set_data_attribute(data_attribute_status, "waiting")
print(f"API to send verification email to {input.email}")
return StateDecision.single_next_state(VerifyState)
class VerifyState(WorkflowState[None]):
def wait_until(self, ctx: WorkflowContext, input: T, persistence: Persistence, communication: Communication,
...
def execute(self, ctx: WorkflowContext, input: T, command_results: CommandResults, persistence: Persistence,
...
class UserSignupWorkflow(ObjectWorkflow):
def get_workflow_states(self) -> StateSchema:
return StateSchema.with_starting_state(SubmitState(), VerifyState())
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.create(
PersistenceField.data_attribute_def(data_attribute_form, Form),
PersistenceField.data_attribute_def(data_attribute_status, str),
PersistenceField.data_attribute_def(data_attribute_verified_source, str),
)
def get_communication_schema(self) -> CommunicationSchema:
...
@rpc()
def verify(
self, source: str, persistence: Persistence, communication: Communication
) -> str:
status = persistence.get_data_attribute(data_attribute_status)
if status == "verified":
return "already verified"
persistence.set_data_attribute(data_attribute_status, "verified")
persistence.set_data_attribute(data_attribute_verified_source, source)
communication.publish_to_internal_channel(verify_channel)
return "done"You can read more details in the iWF wiki. The iWF programming model is really simple to understand, here is a brief summary:
- WorkflowState is the basic unit of async programming. It contains two methods to implement by user code: "waitUntil" and "execute".
- "waitUntil" is invoked by server firstly if implemented, to return some commands for server to wait for. Once commands are completed,
- "execute" is invoked to return a StateDecision.
- The commands can be TimerCommand (as durable timer), or ChannelCommand(wait for some data), or a combination of them.
- StateDecision will be going to next states, completing/failing workflow, etc.
- RPC is for external application to interact with the workflow after it started. Also implement by user code, and invoked by server.
- Persistence is the data that can be accessed in the above WorkflowState/RPC programming.
Since iWF programming model provide "Persistence", it would be much nicer if the persistence schema of the above workflow can be mapped directly to a database scheme, rather than having it as a separate storage outside of database.
User will define the PersistenceSchema like this with additional information to associate with their database:
class UserSignupWorkflow(ObjectWorkflow):
def get_workflow_states(self) -> StateSchema:
...
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.create(
table_name = "user",
primary_key = "id",
PersistenceField.data_attribute_def(col_name=data_attribute_form, Form),
PersistenceField.data_attribute_def(col_name=data_attribute_status, str),
PersistenceField.data_attribute_def(col_name=data_attribute_verified_source, str),
)
...See "table_name", "primry_key" and "col_name".
These mean that that the same workflow code will be operating directly on the database:
- Each workflow execution will be mapped to a row of table "user"
- To map to a specific row, the workflowId is the "primary_key" of the table "id"
- Attributes "data_attribute_form", "data_attribute_status" and "data_attribute_verified_source" are mapped to the columns of the table
And the rest of the code is exactly the same:
- Any
persistence.get_data_attribute(...)is reading from the table row - Any
persistence.set_data_attribute(...)is writing into the table row - The persistence read/write in a WorkflowState/RPC can be atomic if using locking for persistence policy
Though the xdb concepts and SDKs will be mostly the same, they will still have some difference for various reasons.
We will change some concepts based on the feedback/learning of people using iWF, and based on the fact that xdb will be operating directly on database table which usually represent business "objects".
- Workflow --> Process
- WorkflowState --> AsyncState
- WorkflowExecution --> ProcessExecution
- ObjectWorkflow --> ObjectProcess (the interface/baseClass in iWF/xdb SDKs)
- InternalChannel --> InternalQueue
We will drop some functionalities from iWF for now
- Search Attribute: since a table row is mapped directly to a ProcessExecution, users will just use regular database indexes for searching for them. There is no strong need to have search attributes. Though it may come back in the future -- when we need to provide searching based on search engines like ElasticSearch
- SignalChannel: this is a deprecated concept in iWF. Technically InternalChannel/InternalQueue will do the same thing
Something new will/may be added in the future. To keep this proposal simple, they are just briefly described:
- SharedQueue: as opposed to InternalQueue, it's shared across multiple ProcessExecutions
- Read/write with transaction across multiple tables
Please note that, you don't need to understand those new things for the design. They won't be implemented at the first phase of the project.
Some more details will be provided in the FAQ section.
This section will describe the high level design of this project. It will start from xdb users' point of view -- what is the interaction between xdb applications and xdb service. And then go to the architecture of xdb service -- how xdb service interact with its dependencies.
Finally we discuss the actual dependencies that this project will start with.
These are the component names that we will be using:
- xdb worker application: the implementation of process using the xdb programming models, using xdb SDK
- external application: the application that needs to interact with the process: process, stop, waitForCompletion, getResults, invoking RPC etc
- xdb service: the service of xdb as an extension of the user database
- xdb api service: part of an xdb service: the API service handles synchronous request from external applications
- xdb async service: the of an xdb service: the Async service implements the async implementation like invoking AsyncStatewith backoff retry, durable timers, etc
- user database: the user database that xdb will be depending and extending from
- CDC: dependency of xdb service. Change data capture of the user database
- Message Queue: dependency of xdb service. the message queue that xdb will be used for consuming CDC streams, and implementing backoff retry and durable timers
As you may have realized, xdb is essentially a "rewrite" of iWF server, while keep the SDKs and protocol between server and SDKs ("xdb-apis" repo is a folk of "iwf-idl" repo. Same for SDKs).
From user's point of view, they will implement the async process as a worker application which exposed a REST API. XDB service will invoke those APIs. XDB service also expose REST APIs to start/stop/etc processes, which is invoked from external applications that need to interact with the process.
Below is the sequence diagram:
Assuming the process contains a starting AsyncState. So when the process is started, the AsyncState will be executed from xdb service.
The waitUntil API will be invoked if implemented to return some commands. After the commands are completed, then the execute API is invoked to return a decision. Based on the decision, xdb service will continue on -- e.g. execute next AsyncStates.
There two main parts of xdb service: API service and Async service.
The API service is responsible for the synchronous API requests like start/stop a process, from external applications that need to interact with the process.
The API service depends on the user database. For example, when starting a process, it will write some records in the user database, to record that a process is started with an AsyncState to execute.
Then the record of the executing the AsyncState will be delivered to a MessageQueue via CDC. AsyncService will then invoke the worker application's REST APIs(exposed via xdb SDK)
More details to help you understand:
- The user database will have to "install" some xdb table schemas in order to run a xdb service. Like mentioned above, it will need tables for ProcessExecutions, AsyncStateExecutions as "xdb system tables".
- The new xdb system tables will work together with user's other table for their business. XDB will access to both user custom tables for the Persistence API.
- For example, when AsyncService invoking an AsyncState's execute API, it need to provide the requested data attributes. AsyncService will then read from user custom tables. When execute API is returned with some decision to update the data attributes, and also moving to next AsyncStates, AsyncService will atomically update the values in user custom tables, and inserting a new record for a new AsyncStateExecution.
- The new record of AsyncStateExecution will then be delivered to MessageQueue again
- The AsyncService will need to invoke worker application with backoff retry as defined in the StateOptions of the AsyncState
- The AsyncService will need to process the timer command as a durable timer service. There will be a "TimerTask" xdb system tables that deliver the timer task to AsyncService to process.
More details will be provided with example in the "Detailed Design" section.
The end of of this xdb project is to support all the major database products as long as it satisfy the requirements:
- Has transaction or conditional write for writing multiple records in multiple tables atomically
- Has change data capture(CDC) support
As the initial phase of the project, MySQL and Postgres will be implemented first with making sure any other databases can be supported easily .
Apache Pulsar will be chosen as the MessageQueue for the initial phase of the project. Technically, some other message queues can be used as long as it satisfy the requirements:
- Has backoff retry & DLQ on consuming
- Has consumer group for using different cursors on consuming by different consumers
- Has delayed message deliver for building durable timers
For example, SQS + SNS could work as well. Where the visibility timeout can be used for backoff retry + durable timers(need to workaround the 12 hours limit), and SNS can be used for implementing different consumer groups.
One reason of Apache Pulsar is that not only it provides all above required features, but also it has abstraction of CDC out of the box, called "Pulsar IO".