Backend Architecture

ditto's distributed system architecture is a composition of distributed small peers and a big peer cloud deployment ditto uses small peers predominantly for data replication across web, mobile, desktop, and linux embedded systems, such as a raspberry pi, where storage, ram, and cpu resources are generally static and unchangeable conversely, you can think of the big peer as the cloud worker node, which, although viewed as just another peer in the decentralized peer to peer network, is equipped with a replication engine capable of partitioning, sharding, enforcing causal consistency, and more for a complete list of big peer services that you can incorporate into your architecture, see docid\ syt1bx8zpniunkjo7wzse this topic provides a high level technical overview of the main components, properties, and features of the big peer, such as partitioning, sharding, and enforcing causal consistency /#data replication and sync behavior /#key features /#architecture overview data replication and sync behavior the big peer is a cloud based component hosted within ditto's infrastructure that facilitates data sync across different network scenarios; if a small peer has access to the internet, it automatically establishes a connection with the big peer to sync following is an overview of the network scenarios that the big peer facilitates sync for network scenario description mesh networks helps replicate data across peers connected in the same mesh networks, as well as between multiple disconnected mesh networks by way of multihop functionality ( docid wxhye kktuvenblaxkf ) centralized and decentralized systems handles data sync between centralized authorities, such as the big peer; if applicable, your external database, and so on, and decentralized peer to peer networks key features the key characteristics of the big peer include an authentication service that enables connected small peer devices to obtain valid certificates to participate in the mesh and replicate data they have permissions for a data egress service that you can use to safely consume data created and modified within the ditto platform by external services and systems for more information, see docid\ qbwrp9sjxbny ds p es6 a stateless http api for programmatic data ingress for more information, see /#using http api to write offline architecture overview as illustrated in the following graphic, the big peer is a central point of coordination and data storage in the cluster; small peers can sync with the big peer to propagate changes across disconnected meshes, and even back to the centralized server — your enterprise cloud or datacenter component description distributed database the core storage nodes distributed (partitioned and replicated) across the platform these database nodes consume the transaction log; commit data to disk; and exchange metadata, known as background gossip , with other storage nodes ( docid\ fpfqjxlx7lggrbov2fp h ) transaction log a log that captures each transaction, such as upsert, update, remove, and so on ( /#transaction log ) subscription servers specialized nodes that extend big peer functionality by bridging communication between all peers and various data sources, as well as serve as caches to enhance query performance ( /#subscription servers ) the following graphic provides a high level overview of the orchestration between the subscription server, transaction log, and distributed database distributed database although the big peer appears like any other peer in the peer to peer mesh network, internally the big peer is a distributed database capable of partitioning and replicating data for scale, performance, and fault tolerance for more information, see docid wxhye kktuvenblaxkf the distributed database is comprised of core storage nodes distributed across the network each storage node runs a local rocksdb open source embeddable key value database that stores documents and is capable of automatic indexing automatic indexing is a query optimization and data retrieval mechanism that creates, manages, and utilizes the indexes that are distributed across storage nodes in addition, each storage node has a mapping of partitions to servers, known as an in memory cluster configuration horizontal scaling in a kubernetes environment, as long as the big peer has persistent volume capacity to attach a hard drive to the running storage server process, referred to as the stateful sets , the big peer scales horizontally the details of a cluster — including its size, shape, members, partitions, replicas, and more — all reside within a cluster configuration when a cluster needs to scale up and down, ditto creates a new cluster configuration and instructs the big peer to transition away from the current configuration to the new one when reconfiguring a cluster, there is minimal data movement that is, if the cluster grows, the objective is to move only the minimum amount of data necessary to accommodate the new nodes subscription servers the subscription servers act as soft state satellite api nodes for receiving and distributing small peer data by functioning as an api gateway in the cluster that is, the subscription servers bridge the peer to peer mesh network and the centralized datacenter or cloud summary as an api gateway, subscription servers are responsible for the following essential functions provide a bridge for small peers to read and write data to the big peer, including managing the intricacies of replicating data across peers — small peers, as well as small peers and the big peer manages authentication, permissions, and document caching if upgraded to the kafka connector, s erves as the producer of transactions for the transaction log s — when the subscription server receives data from a small peer, it creates a transaction that contains all of the data from a round of replication and pushes it onto the kafka log for more information, see docid\ qbwrp9sjxbny ds p es6 working alongside ditto's http api, ensures that, regardless of network conditions and internet availability, peers maintain an accurate and up to date view of the data soft state satellite api nodes when a small peer initially connects to a subscription server, the subscription server triggers a query to determine the current distributed database state soft state nodes play a crucial role as temporary repositories for locally cached data in scenarios where cached data is comprehensive for a small peer's subscription, querying the store nodes might not be required these specialized nodes, functioning as caches, extend the big peer's functionality by bridging communication among all peers and diverse data sources additionally, they serve as a means to enhance query performance, leveraging routing keys and data locality consequently, the subscription server, when appropriate, “pushes” that data to subscribing peers partitioning and replication the primary role of the big peer is to store datasets that are too large to fit on any one individual small peer device and maintain data synchronicity between all database replicas running locally within small peer end user environments, such as an iphone mobile device from the outside, the big peer appears as a single node; however, internally, the big peer runs as a distributed database divided into partitions each partition consists of a single node in the kubernetes cluster if a collection contains many documents stored in a single partition, a single node responds to all of the queries, which has the potential to significantly reduce query performance using http api to write offline in serverless environments where you have cloud service reliability of at least 99 99% uptime and therefore high internet access probability, you can use ditto's http api to write data from your external system, such as a third party datacenter or cloud account, to offline small peers when internet connection is dependable and consistent, instead of needing to implement processes like replicating data across the network or caching data to local small peer environments in your code, the subscription server facilitates those intricate processes for you so even when small pees are offline, you can operate under the assumption that your update invoked using the http api is consistently available across the platform, even when small peers are offline for more information, see docid 9hxnrg96qjpfsmta8vrvg managing local memory garbage collection the big peer periodically runs the garbage collection (gc) process in an effort to reclaim memory and improve performance by automatically identifying and purging data that is no longer relevant or useful the big peer uses the gc timestamp to determine which document versions are candidates for purge for more information, see docid\ fpfqjxlx7lggrbov2fp h managing queries read transactions managing reading and read transactions in distributed systems is complex since you need to consider factors like consistency, isolation, and how to resolve concurrency conflicts the coordinating node is responsible for managing and coordinating queries within the cluster each distributed soft state storage node maintains a local copy of the partition map retrieved from the cluster configuration a partition map is a specialized data structure used for managing the propagation of data across partitions within a given cluster this setup allows any storage node within the cluster to potentially assume the role of a coordinating node when needed coordinator node the coordinator assesses the query to determine which partitions contain the data necessary to answer the query as follows the coordinator compiles a list of required partitions once compiled, the coordinator then selects one replica from each partition based on a fault detection mechanism, choosing the replica least likely to have faults the query is sent to each replica, and the results are merged and streamed back to the requester for a technical overview of failure detectors, see the academic paper " https //www researchgate net/profile/benjamin satzger/publication/4339407 a lazy monitoring approach for heartbeat style failure detectors/links/0912f50bcb4b32fc08000000/a lazy monitoring approach for heartbeat style failure detectors pdf " by the university of augsburg random slicing https //hpc ac upc edu/pdfs/dir05/file004529 pdf has been written about brilliantly in https //www infoq com/articles/dynamo riak random slicing/ by scott lystig fritchie, which motivates the why of random slicing as well as explaining the how here, we will briefly discuss big peer's implementation we made a decision to make this first version of big peer as simple as possible, and so we elected to keep our cluster shape and replica placement very simple (though it is extensible and will get richer as time allows or needs dictate) each document in big peer has a key, or document id which is made up of a namespace (the application (appid) and the collection) and an id for the document our current hashing policy involves hashing the entire document id, ensuring that data within the same collection resides in the same partition as per the random slicing algorithm, we think of the keyspace as the range 0 to 1 we take the capacity of the cluster, and divide 1 by it this determines how much of the keyspace each partition owns in our initial implementation the capacity is the number of partitions we wish to have we enforce an equal number of replicas per partition, and thus all clusters are rectangular e g 1 1, or 2 3, or 5 2, etc , where the first number is the number of partitions, and the second the number of replicas random slicing allows in future to have heterogeneous nodes, assigning the capacity accordingly in the case that we want three partitions of two replicas, we say each partition takes 1/3 of the keyspace, or has 1/3 of the capacity hashing a documentid then gives us a number that falls into the 1st, 2nd or 3rd 1/3 of the keyspace, and that decides which partition owns that document we can transition from any configuration to any other, and we do this by slicing or coalescing partitions using the cut shift algorithm from the random slicing paper the graphic below illustrates how this looks as the image shows, partition 4 (p4) is made up of slices from p1, p2, and p3, these three slices we call intervals they represent, in this case, two disjoint ranges of the keyspace that p4 owns a replica of p4 has two intervals, whereas p1 has a contiguous range and a single interval our random slicing implementation is currently limited in that resources must be added and removed in the cluster in units equal to the desired replication factor if you want to add a node, and your desired replication factor is two, you must add two nodes this is not a limit inherent in random slicing, but a choice we made to speed up implementation as scott's article points out, random slicing matches your keyspace to your storage capacity, but that is it! it doesn't manage replica placement more complex replica placement policies are coming, read scott's article in short, random slicing appears very simple, map capacity to the range 0 1, and assign values to slices in that range cut shift is a great way to efficiently carve new smaller, partitions from slices of larger ones, and coalesce smaller slices into larger partitions when big peer scales up or down each storage node uses the random slicing partitioning information to decide if it needs to store documents from any given transaction if the random slicing map says that server one owns documents in the first partition, then for each transaction server one will store documents whose ids hash to the first partition interval maps, missed transactions, and backfill​ each storage nodes keeps a local data structure, stored durably and atomically with the document data, that records what transactions the node has observed the structure is called the intervalmap , and represents what has been observed, in what slices of the keyspace for example, if a server is responsible for an interval of the keyspace that represents the first third of the keyspace, the server "splices" the observed transactions into the intervalmap at that interval imagine server 1 is responsible for interval 1, it receives transactions 1 =100 from the log, it adds the data from those transactions to a local write transaction with rocksdb then it splices the information into the intervalmap, that it has seen a block of transactions from 1 =100 we now say that the base for interval 1 is 100 now the server stores this updated intervalmap with the data in a write transaction to rocksdb next the server receives transaction 150 =200 from the log clearly the server can detect that it has somehow missed transaction 101 =149 the server can still observe and record the data from these new transactions, and splice the information into the intervalmap the intervalmap now has a base of 100 and a detached range of 150 =200 any server with any detached ranges can look in the partition map to see if it has any peer replicas, and ask them for the detached range(s) this is an internal query in big peer if a peer replica has some or all of the missing transaction data, it will send it to the requesting server, who will splice the results in the intervalmap , and write the data to disk this way a server can recover any data it missed, assuming at least one replica stored that data we call this backfill nodes gossip their intervalmaps , this is how the ust is calculated, and how backfill replicas can be chosen read on down to "missed/lost data" if you want to know how the cluster continues to make progress and function in the disastrous case that all replicas miss a transaction the intervalmap , gossip, backfill, ust, read transactions, and the gc timestamp all come together to facilitate "transitions", which is how big peer can scale up and down, while remaining operational, available, and consistent transitions also mentioned in scott's article on random slicing is the fact that random slicing will not tell you how, or when, to move your data around if you want to go from one set of partitions to another in big peer we have the added problem that we must at all times remain causally consistent big peer manages transitions between configurations by leaning on those two primitives the ust and the gc timestamp the process is best explained with an example using the diagram from the random slicing section, a walkthrough of the transition from the three partition original cluster to the target four partition cluster in this case assume two replicas per partition, which means adding two new servers to the cluster there is a current config, that contains the intervals that make up the partitions 1, 2, and 3 mapped to the replicas for those partitions the name p1r1 refers to the first replica of partition 1, p2r2 the second replica of partition 2, etc in the current config there are nodes p1r1 , p1r2 , p2r1 , p2r2 , p3r1 , p3r2 two new nodes are started, ( p4r1 , p4r2 ) a new cluster configuration is generated from the current configuration this runs the cut shift algorithm and produces a next configuration, with the partition map and intervals as per the diagram above we store the current config, and the next config in a strongly consistent metadata store updating the metadata store causes the current config and next config file to be written to location on disk for each big peer store node, and each node is signaled to re read the configs the servers in p1 p3 are all in the current config, and the next config the servers in p4 are only in the next config a server will consume from the log if it is in either config those in both configs will store data in all intervals they own in both configs in our example each of the current config servers stores a subset of the current sub interval of its current ownership in the next config the new servers in p4 start to consume from the log at once, and gossip to all their peers in both configs backfill (again) for example, we start the new servers when the oldest transaction available on the log is txn id 1000 they must backfill from 0 1000 from the owners of their intervals in the current configuration they use the current configuration to calculate those owners, and the intervalmaps from gossip to pick an owner to query for data no longer on the log recall that the ust is calculated from the base of the intervalmap but these new servers (only part of the new config) do not contribute to the ust until they have backfilled routing and ust in the section on ust we described a scalar value, the transaction timestamp in reality this value is a pair of the configurationid and the ust the configurationid rises monotonically, the initial configuration being configid 1, the second configid 2, etc this allows us to calculate a ust per configuration before we began the transition the ust was (1, 1000) the ust may never go backwards (that would break causal consistency) after starting the new servers and notifying nodes about the next config, the ust in the current config is (1, 1000) and in the next config is (2, 0) during this period of transition the nodes in p4 cannot be routed to for querying only nodes in the current config can coordinate queries, and these nodes decide what configuration to use for routing based on the usts in each of the current and next config we call this the routing config it is calculated and like everything else in big peer, it progresses monotonically upwards after the new nodes have backfilled, and after some period of gossip, the ust in the next config arrives at a value that is >= the ust in the current config so the servers in the current config will begin to route queries using the next config recall that nodes gossip a gc timestamp that is based on active read transactions a read transaction is identified by the timestamp at which it began for example (1, 1000) is a read transaction that began at ust 1000 in the current configuration when all the replicas in the current configuration are routing with the next configuration, (e g , the cluster gc timestamp is in the next configuration, (2, 1300)) the transition is complete any of the nodes can store the next config into the strongly consistent metadata store as the current config each node is signaled, and eventually all will have a current config with configid 2, and will forget metadata related to configid 1 furthermore, garbage collection will ensure that replicas drop data that they no longer own transaction log as a core part of ditto, transaction logs capture information about changes made to the local ditto store for each transaction, including the type of transactional operation performed, such as upsert, remove, and so on enforcing causal consistency the big peer’s replication engine uses this transaction log to enforce causal consistency by identifying each transaction’s sequence and execution time as follows if upgraded to the kafka premium option, the big peer utilizes the kafka commit log as its underlying data structure for storing and processing messages from your kafka queue for more information, see docid\ qbwrp9sjxbny ds p es6 and docid\ zqee8xarjtm0gjpsjbhwy when the transaction is committed, the sequence number represents the current version of the database in the transaction log, a transaction’s execution sequence number becomes the transaction timestamp , which reflects the ust (timestamp) for more information, see docid\ hjshz87ybyk1f65jfx0v7 the logged transaction timestamps form a total sequence from zero and above in this context, "zero" refers to the initial state of the datastore, in which no data is present each storage node observes data from the log once the sufficient number of nodes, as defined by the configurable quorum , have observed and agree to the data in the transaction log, the transaction is considered valid and written durably that is, the transaction is written in a way that ensures consistency, integrity, and reliability throughout the system gossipping for high availability in decentralized communication, each node gossips its highest transaction committed background gossip refers to the communication between nodes to share information about their state without relying on a centralized authority the gossiped data is encapsulated in an announce message, which contains the announcing server's intervalmap , gc timestamp, highest observed clusterconfig epoch, and a boolean that indicates if the server has backfilled in that clusterconfig epoch the intervalmap is a mapping from ranges of document ids, referred to as intervals , to transactionsets the intervalmap expresses exactly which transactions have been applied in which intervals from this gossip, individual nodes calculate what they consider to be the ust if every server gossips its local maximum committed transaction , then the ust is the minimum of those maximums the local maximum committed transaction refers to the process in which a node shares information about the highest transaction it has committed to its local storage put differently, the ust represents a limit that should not be exceeded — the “upper limit ” for more information, see docid\ hjshz87ybyk1f65jfx0v7