XDB Proposal & Design
This is the design and proposal doc of xdb project. It will starts with the goal of the project, then the high level design with tech stacks, and how all the components work together. Then a detailed design section with example.
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", primary_key_type = String,
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 mostly 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, and the xdb system tables for the rest.
- 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 goal of of the xdb project is to support all the major databases 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".
This sections will provide a lot more details of how to implement the xdb service, with an example to walk through how each components work together.
Note these are just for design discussion purposes. Just like any system designs, there could be some modification/difference in the actual implementation.
For iWF, a "workflowId" is needed when starting a workflow. And the same workflowId can be reused to start different WorkflowExecutions. It's called "IdReusePolicy". A "RunId" is returned from server to identify a WorkflowExecution.
This "workflowId" and IdReusePolicy is very powerful/useful. WorkflowId can be used directly to represent business identifier. For example, "user-signup-userId123" can be the workflowId for "userId123". When the workflow execution failed for any reasons, another workflow execution can be started to reuse the workflowId.
Similarly in XDB, a "ProcessId" is needed to start a ProcessExecution. To reuse the same ProcessId, a ProcessExecutionId will be returned.
In addition, because the persistence of a ProcessExecution is mapped to a table row, it needs additional info like tableName, primary_key_value. primary_key_value is needed so that xdb service can know which row of the table to load as the persistence.
The code to start a process will be like below
client.start_object_process(
"user-signup-process-userId123", # processId
UserSignupProcess, # Process class
"userId123", # objectId for primary key
input, # input for stating state if applicable
ProcessStartOptions(...) # other options like IdReusePolicy, initial data attributes
)As mentioned above, to extend the user database, some system tables are needed to implement the programming model.
Below are the system tables that will be required:
| Column | Type | Description |
|---|---|---|
| id(PK) | string | the process execution id |
| process_id | string | the process id |
| is_current | bool | whether or not it's the current/latest execution for the process_id |
| status | string | running/timeout/completed/failed |
| start_time | datetime | the start time |
| timeout_seconds | int | the timeout |
| info | blob | store additional static info like processType, workerUrl, tableName and primary_key value, use blob for extension |
Secondary indexes:
- (process_id, is_current)
Because of "IdReusePolicy" of ProcessId, "is_current" is needed to know which is the latest/current execution for the same ProcessId.
| Column | Type | Description |
|---|---|---|
| process_execution_id(PK) | string | the process execution id |
| async_state_id(PK) | string | the state id |
| state_execution_number(PK) | int | the number to indicate this is Xth execution of the async_state_id |
| status | string | running/completed/timeout/fail |
| input | blob | input of the state execution |
| info | blob | static info includes StateOptions |
| wait_until_status | string | skipped(if waitUntil API is skipped), running, waiting_commands_completed, completed |
| wait_until_commands | blob | the command details that will be waiting for |
| wait_until_command_results | blob | the command completed results that have got so far |
| execute_status | string | not_started, running, completed |
| last_state_api_invoked_time | datetime | last time that invoking the state APIs(waitUntil/execute) for backoff retry |
The primary key is composed by three columns: process_execution_id, async_state_id and state_execution_number. They together uniquely identify a state execution.
This is a critical table that CDC/MessageQueue/AsyncService will be using.
When a new record is inserted, CDC/MessageQueue will deliver it to AsyncService, which will then start invoking the State APIs for user worker applications.
| Column | Type | Description |
|---|---|---|
| id(PK) | string | the timer id |
| firing_time | datetime | the firing time |
| timer_type | string | process_timeout or timer_command |
| info | blob | additional info like process_execution_id, state_execution_id |
This is a critical table that CDC/MessageQueue/AsyncService will be using.
When a new record is inserted, CDC/MessageQueue will deliver it to AsyncService, which will start a "delayed message" for durable timer. When the timer fire, the AsyncService will invoke the API service for processing the timer firing event.
The "wait_until_commands" in "xdb_sys_async_state_execution" will be referring to this table if there is a "TimerCommand". This means that when "waitUntil" API returns a timerCommand, API service will update the "xdb_sys_async_state_execution" and also inserting a new record for this xdb_sys_timer_tasks.
Similarly, to implement the process timeout for the whole process execution, API service will also insert a new timer record when inserting a "xdb_sys_process_execution".
| Column | Type | Description |
|---|---|---|
| process_execution_id(PK) | string | the process execution id |
| queue_name(PK) | string | the name of the queue |
| sequence_id(PK) | int | the auto increment id for FIFO sequence of the queue |
| message_id | string | an id for deduping messages received from external applications |
| message | blob | the value of the message |
Secondary unique index:
- (process_execution_id, queue_name, message_id)
This table represents the InternalQueue, which stores the messages for the queue. API service will write messages into this table, and also checking if the process_execution has met the required InternalQueueCommands.
Here uses this "User Signup Workflow" as an example, show the worker and external application code, and how they execute through the XDB service, with details of how the above xdb system tables work together.
The requirements are copied from the iWF samples "User Signup Workflow" :
- User fills a form and submit to the system with email
- System will send an email for verification
- User will click the link in the email to verify the account
- If not clicking, a reminder will be sent every X hours
For this use case, user just need a "user_sign_up" table:
| Column | Type | Description |
|---|---|---|
| user_id(PK) | string | the user id |
| form | blob | the other static user info that submitted from sign-up form, use blob for simplification |
| status | string | the status of this user account: "waiting" or "verified" |
| source | string | the source of where this is verified at |
The worker application code is mostly the same as iWF, with slightly different at the PersistenceSchema, like we have mentioned above:
class SubmitState(AsyncState[None]):
def execute(self, ctx: WorkflowContext, input: Form, command_results: CommandResults, communication: Communication,
) -> StateDecision:
input = persistence.get_data_attribute("form") # the initial data attributes will be set when starting the process, as like https://github.com/indeedeng/iwf/issues/278 so that database is source of truth before sending emails
print(f"API to send verification email to {input.email}")
return StateDecision.single_next_state(VerifyState)
class VerifyState(AsyncState[None]):
def wait_until(self, ctx: WorkflowContext, input: T, communication: Communication,
) -> CommandRequest:
return CommandRequest.for_any_command_completed(
TimerCommand.timer_command_by_duration(
timedelta(hours=24)
),
InternalQueueCommand.by_name("verify"),
)
def execute(self, ctx: WorkflowContext, input: T, command_results: CommandResults, persistence: Persistence,
communication: Communication,
) -> StateDecision:
form = persistence.get_data_attribute("form")
if (
command_results.internal_queue_commands[0].status
== QueueCommandStatus.RECEIVED
):
print(f"API to send welcome email to {form.email}")
return StateDecision.graceful_complete_workflow("done")
else:
print(f"API to send the a reminder email to {form.email}")
return StateDecision.single_next_state(VerifyState)
class UserSignupProcess(ObjectProcess):
def get_workflow_states(self) -> StateSchema:
return StateSchema.with_starting_state(SubmitState(), VerifyState())
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.create(
table_name = "user_sign_up"
primary_key = "user_id", primary_key_type = "string",
PersistenceField.data_attribute_def(col_name = "form", Form),
PersistenceField.data_attribute_def(col_name = "status", str),
PersistenceField.data_attribute_def(col_name = "source", str),
)
def get_communication_schema(self) -> CommunicationSchema:
return CommunicationSchema.create(
CommunicationMethod.internal_channel_def("verify", None)
)
@rpc()
def verify(
self, source: str, persistence: Persistence, communication: Communication
) -> str:
status = persistence.get_data_attribute("status")
if status == "verified":
return "already verified"
persistence.set_data_attribute("status", "verified")
persistence.set_data_attribute("source", source)
communication.publish_to_internal_queue("verify")
return "done"The code to start a process will be like below
client.start_object_process("user-signup-process-"+user_id, UserSignupProcess, user_id , None, # None because the startingState doesn't need input
ProcessStartOptions(
initial_upsert_data_attributes = {
"form": form_input,
"status": "waiting",
}
)
)With the above setup(user database, worker and external application code), here is how they will be executed internally in XDB service:
This request is sent to the API service of a XDB service. The API service will atomically(using database transaction):
- Insert a new record into
xdb_sys_process_executionif not exists - Insert a new record into
xdb_sys_timer_task - Because
UserSignupProcesshas a starting state, API service will also insert a new record toxdb_sys_async_state_execution- SubmitState doesn't implement
wait_until, so thewait_until_statuswill be "skipped"
- SubmitState doesn't implement
The AsyncService will receive two events because of the above transaction in the user database:
- A new timer
- A new AsyncStateExecution
For the new timer, AsyncService will produce a new "delayed message" back to Plusar to wait for the timer to fire. When the timer fires, AsyncService will call back to API service to process the timer. Here omits this timeout flow, to focus on the happy case. It should be straightforward to implement the timeout in API service.
For the new AsyncStateExecution, AsyncService will make a callback to worker application to execute the SubmitState.execute API implementation. The worker implementation will then send an email to user. The callback is controlled with backoff retry which can be easily done with Pulsar.
Note that before AsyncService invoke the State APIs, it will load the data from user tables according to the PersistenceLoadingPolicy.
SubmitState.execute returns a StateDecision to go to next state. On receiving this StateDecision, AsyncService will atomically:
- update the status of the current AsyncStateExecution
- write a new AsyncStateExecution for the next state
VerifyState
The new AsyncStateExecution for VerifyState again will go through CDC->Pulsar and finally delivered to AsyncService as another event.
This time, the state needs wait_until API. So AsyncService will call back to worker application VerifyState.wait_until to execute the API implementation. It will return:
CommandRequest.for_any_command_completed(
TimerCommand.timer_command_by_duration(
timedelta(hours=24)
),
InternalQueueCommand.by_name("verify"),
)
To handle this commandRequest, AsyncService will atomically:
- check if the
InternalQueueof "verify" has any message to complete the command. - if there is no, it will then mark the status to "waiting_commands_completed" with the commandRequest
- it will also write a new record to the xdb_sys_timer_tasks for the TimerCommand of 24 hours
AsynService will receive an event for the new timer task that just added. Similarly like last time, it will produce a "delayed deliver message" to Pulsar for waiting the timer to fire.
Assuming nothing changes for 24 hours, so the timer fires as the "delayed delivered message" arrives. AsyncService will receive this delayed message, and call API service for this timer firing
Now API service will check if this timer firing is enough to complete commandRequest of the stateExecution. If not enough, it will continue to wait.
In this case, it's enough -- because the commandRequest is "any_command_completed(...)".
Therefore, it will:
- mark the wait_unti_status as "completed", and "execute_status" as "running"
- add the timer fired to the "wait_until_command_results"
- clean up the timer tasks
On receiving event that an AsyncStateExecution is "running", AsyncService will then invoke the VerifyState.execute API with the commandResults. The commandResults contain information that the timer has fired, but no message received for the InternalQueue.
Therefore, worker application will go to the branch to send a "Reminder Email" to user, and then return a StateDecision to go back to the VerifyState again.
On receiving the StateDecision, AsyncService will mark the current StateExecution as completed, and insert a new record for AsyncStateExecution.
Then, the same thing will happen again like in "CDC->Pulsar->AsyncService 2" and "CDC->Pulsar->AsyncService 3" to execute the new AsyncStateExecution's wait_until and start at timer.
This time, the signup user clicks a link in the email, which will invoke an RPC as external application:
client.invoke_rpc(username, UserSignupWorkflow.verify, source)When the API service handles this request, it will invoke the RPC API implementation in worker application.
Similar to AsyncService invoke the State APIs, before invoking RPC, API service will load the data from user tables according to the PersistenceLoadingPolicy.
The UserSignupProcess.verify will be invoked and return a response:
- update the data attribute "status" to "verified" " update the data attribute "source" to "email" " publish a message to the InternalQueue "verify".
This RPC response will be processed by API service, atomically:
- update the "user_sign_up" table columns "status" and "source"
- check if the new message to be added to InternalQueue "verify" can be enough to complete a stateExecution
- if not, then just add it to the queue
- if yes, then add the message to the "wait_until_command_results", mark the wait_until_status to completed and execute_status to running
On receiving event that an AsyncStateExecution is "running", AsyncService will then invoke the VerifyState.execute API with the commandResults.
This timer, the commandResults contain information that the InternalQueueCommand has completed. So it will send an "welcome email" to signup user, and return a StateDecision to complete the process.
On receiving the StateDecision, AsyncService will mark the stateExecution as completed, and the processExecution as completed.
TODO
TODO
- Search Attribute
- Caching