Decentralised Identity 🤝 Mesh Networking
How you can invite your Bluesky friends into your Tailscale network using an OIDC shim
Increasing lot sizes of our digital homes
From GeoCities to Facebook to Open Social and beyond: a history of digital homes, and a thought experiment for what comes next.
Rabbithole
Tabbed browsers have confined us internet users to a reduced representation of our engagement with content online. We are bound to a linear, sequential history with tabs when what we're doing on the internet much better resembles a journey through a sea of information.
[From Medium] Distributed Systems II: Leaderless and P2P Systems
By Govind Mohan and Sajeed Syed Bakht Note from Gov: this is a 2 part article I wrote with a friend on Medium in 2020 In our previous article on distributed systems, we explored the leader-follower paradigm where write requests are handled by a leader and database state updates are subsequently propagated to followers, which handle reads. We saw a drawback of these systems in that they are more suited to high read throughput and lower write throughput. In this article, we will check out systems that don’t have a leader. Having a leader is typically useful if consistency is a key system requirement. However, it may be more important for some systems to be highly available, which means any node returns a non-error response to a query from our definition in the previous article. In such cases, a leaderless system is better suited. Leaderless systems, as the name suggests, do not have leaders. All nodes (often referred to as peers) are identical in function. As a result, they do not have failover mechanisms for leader loss as there is no need to replace leaders. Instead, measures need to be put in place to ensure the distributed database state achieves eventual consistency. Thus, these systems typically have a background process to synchronize state between nodes. This process of information exchange between peers is known as gossip. Intuitively, the idea of gossip is a very good fit for describing communications in leaderless systems. Gossip in the human context could be the act of spreading rumors, and as the saying goes, “anyone can start a rumor, but none can stop one”. While this is dubious in the human context, it’s a very desirable property for leaderless systems as there are no guarantees that a node will remain live for an extended period of time. Further, all the peers run the same program but none of them are singularly responsible for orchestrating how reads/writes are propagated across the system. Hence, they need to gossip the information (writes) they receive with their neighbors. We will explore two types of leaderless systems here: Dynamo-based systems, and P2P systems. Dynamo-based Systems Amazon’s internal database system is called Dynamo, which follows the leaderless paradigm and has inspired various similar systems. Interestingly enough, Amazon Web Services offers a similarly named DynamoDB as a service, which is actually a single leader system. Our focus here is on the former. A key feature in Dynamo-based leaderless systems is that read and write requests to the system are sent in parallel to multiple peers. This way, if some peers are offline others can receive the request. Offline peers can be updated when they come back online. However, if an offline peer misses a write and comes back online then it may not get a chance to synchronize new writes with its neighbors. Thus, if a peer receives a read at this state it will provide a stale value. Peers can track stale values by maintaining a version number along with each record, which they return in read responses and update with modifications to records. Synchronizing Mechanisms If queries are spread across the network in parallel, there needs to be a synchronizing mechanism to ensure the system attains eventual consistency. Dynamo-style leaderless systems have two such mechanisms. 1. Read repair Since a read request gets versions from different peers in the network, it can find the most recent version and all the versions that differ from it. Thus, the client can find the latest version among all responses to its query, and send it to all peers that sent an outdated version. These peers will subsequently update their stale values. 2. Anti-entropy Dynamo inspired systems can also have a background process running in each peer that looks for discrepancies in its datastore by frequently querying other peers and looking for outdated values. Quorum We can notice that eventual consistency is possible using synchronization mechanisms, however the system needs to ensure that a write successfully occurs in enough peers so that it is accessible to future reads. This relationship can be described by the equation w + r > n, where w is the number of writes, r is the number of reads and n is the number of replicas. This means that the number of reads and writes have to be greater than the number of replicas as reads are guaranteed to reach peers where the write was successfully registered. This is known as quorum, as writes and reads must be able to agree on whether a write is confirmed in the network. P2P (Distributed Hash Table) Peer-to-peer, or P2P systems are leaderless networks which have a content addressing system. In other words, peers do not maintain full copies of the database, rather, records are distributed and replicated individually as key-value pairs with each peer containing a Distributed Hash Table (DHT) to find content in the network. As a result of records being distributed across the network, it does not require a quorum mechanism like Dynamo networks to ensure records are available. The DHT on a peer stores locations for records so they can be found across the network. This way, if a peer gets a read, it refers to its DHT to fetch the data from the correct peer. It is thus important that writes to the network are propagated far enough that any peer in the network is able to reach it, and also that all the network peers can still respond to reads if some peers have faults. There are various DHT protocols, such as the following, that accomplish this. They all share the approach of defining a virtual structure (like a circle or a tree) as an overlay network on top of the original network in order to make reads/writes efficient. Chord In this protocol, all peers are assigned an address that is of size m bits. The addresses are arranged from 0 to 2ᵐ as points in a circle like the diagram below. Records that are stored in the network are also assigned a key that is m bits long. A hashing algorithm such as SHA1 is used to generate these IDs as they need to be collision resistant (the same key or address shouldn’t point to two different values or peers) for the system to work. A record’s key is numerically close (greater than/less than/equal) to a peer’s address, since they both occupy the same number of bits. In fact, write queries work by generating an m-bit key for the write contents and storing the key-value pair in the peer with the ID closest to that key. “Closest” can be defined to mean greater than or less than by convention. Peers are randomly assigned a peer ID and thus the Chord circle can have gaps in it. As a result, a write with a key k is written to the peer with ID defined by successor(k) which returns the ID of an active peer with the closest address to k. image This addressing scheme is used in the DHT for finding content. A basic read occurs when a peer receives a query for a key, it passes the query on to the next peer (in the address circle) unless it contains the key in its datastore. However, this takes O(n) time which is quite costly in large networks. There is a faster solution that involves each peer maintaining a “Finger Table”. This table contains up to m ordered entries. To see this in action, if the node with ID N8 receives a query for key with ID K53, it will refer to its finger table for the node with the closest ID to 53 that it is aware of. From the diagram, we can see that this is N42. Thus, N8 will forward the request to N42. This node will see that N51 is the closest ID it is aware of less than 53. Finally, N51 will forward the request to N56 which will return the value for K53 (as there is no node closer to 53 counting upwards than N56). More generally, the finger table entry at row i will contain the peer with the address closest to the next power of 2 offset by the distance between the current peer and the peer with address 0. This can be represented in fancy notation (with n being the current peer’s address) as: image This modification makes reads happen in O(log(n)) as the query is passed to the finger table entry for the closest successor to the key specified in the query, which by design is spaced out by 2ⁿ. Kademlia Kademlia operates on very similar ideas to Chord with a few key differences. For starters, both peers and records in a Kademlia network maintain the property of being addressed in the same m-bit space. The key difference is that it uses a binary tree as an overlay network rather than the circle topology of Chord. image Visualizing this isn’t easy so let’s refer to the above diagram. Nodes and keys are addressed by a binary trie, which is a form of addressing using prefixes of a binary sequence. This means any path from the root node to a leaf will represent a unique address attained by concatenating the edge values. At any node the right forward edge will have the value 0 and the left forward edge will have the value 1. Let’s focus on the leaf node on the right side with all the curved arrows coming out of it. The address of that node will be 0011, which is the path from the root node to it — root, right, right, left, left. This is a pretty elaborate scheme so it would make sense that there is some benefit to it. In fact, this addressing scheme has the massive benefit of being able to determine the distances between addresses (either a peer or a record) using the XOR operation. image Two questions arise at this point: where did XOR come from? Why is it even relevant? It turns out that XOR captures the notion of distance implicit in this binary tree structure. As an example, notice that 1001 XOR 1101 is 0100. Let’s tie this back to our tree structure; in a fully populated tree of m-bit IDs, the distance between two IDs is the height of the smallest subtree containing both of them. Referring back to our binary example, 0100 shows that the height of the smallest subtree with both is 4. As to why the XOR metric is important, it maintains the notion of distance very well. In Chord, for example, a node at the first quadrant will think of a node in the second quadrant as ‘close’ since it needs few successive hops. However the node in the second quadrant will deem the node in the first quadrant as ‘far’ since it has to hop all the way across the third and fourth quadrants to get to the first quadrant. XOR thus works much better as a distance metric given that it works both ways (i.e. a XOR b = b XOR a). Now let’s look at how queries work in Kademlia. Each peer maintains a list of contact information for other nodes, typically IP address, UDP port, Node ID (as Kademlia messages are passed using UDP). Further, this list is divided into several sublists known as k-buckets. The i-th k-bucket stores k nodes with IDs between 2^i and 2^{i+1} from itself. This can be visualized as follows: image As you might notice from this structure, k is fixed while the number of possible nodes between 2^i and 2^{i+1} is strictly increasing. Thus the k-bucket structure ensures that a peer knows a lot about its neighbors and less about nodes that are far away. Now let’s tie all this together to see how a query works. A peer receiving a query for a node with ID m will refer to the k-bucket it has that contains IDs closest to m by XORing m with its own ID. As mentioned earlier, this will provide the height of the smallest subtree that contains both IDs. This can be used to determine which bucket m belongs in. Since it only knows at most k peers in this bucket, it will forward the request to some of those nodes (the exact number is set as a system-wide concurrency parameter). This is done simultaneously, as some of those nodes might have failed in the time since their last communication with the node sending the message. Any of those nodes that is aware of the node that contains the value for the key m will forward the query to the corresponding node and send an acknowledgment to the original querier. Since each k-bucket essentially stores contacts from various subtrees as in the above diagram, the query will converge logarithmically to the correct node. In other words, a Kademlia network with 10,000,000 nodes would only require at most 20 hops for any node to match a message ID to its value! There are several very interesting and complex ideas in the world of leaderless systems. Peer-to-peer networks specifically have several extremely valuable properties because they offer the possibility of fully trustless networks. In future articles, we will go through what trustless networks actually entail by examining concepts such as consensus, byzantine fault tolerance and cybersecurity in p2p systems. They are a waning presence on the internet as we are moving to a more cloud centric approach, however there are still some pioneers in the p2p world such as IPFS and Virgil Systems, who are redefining the internet from the ground up in a trustless, decentralized manner for data/content management. Citations - Petar Maymounkov and David Mazières. 2002. Kademlia: A Peer-to-Peer Information System Based on the XOR Metric. In Revised Papers from the First International Workshop on Peer-to-Peer Systems (IPTPS ‘01). Springer-Verlag, Berlin, Heidelberg, 53–65. - Ion Stoica, Robert Morris, David Karger, M. Frans Kaashoek, and Hari Balakrishnan. 2001. Chord: A scalable peer-to-peer lookup service for internet applications. SIGCOMM Comput. Commun. Rev. 31, 4 (October 2001), 149–160. DOI:https://doi.org/10.1145/964723.383071 - Kleppmann, M. (2019). Designing data-intensive applications the big ideas behind reliable, scalable, and maintainable systems. Beijing: O’Reilly.
[From Medium] Distributed Systems I
By Sajeed Syed Bakht and Govind Mohan Note from Gov: this is a 2 part article I wrote with a friend on Medium in 2020 The world of databases can be difficult to navigate whether you’re a student, a fresh graduate, or a seasoned developer. Several paradigms have emerged over the last decade, all embodied by various large projects with eccentric names like Cockroach DB and Voldemort. We are also experiencing a new wave of innovation in distributed systems with Distributed Ledger Technology which was popularized by the rise of cryptocurrencies such as Bitcoin and Ethereum. In this article we’ll explore the major paradigms, and provide some context as to why these paradigms are popular and what has led to their popularity. The Need for Replication A database that runs on one server can satisfy all these cases, but as traffic increases (a natural side-effect of growth), it loses its ability to be fault tolerant, that is to perform without errors. There can be issues with memory/disk/CPU limitations, and hardware upgrades (vertical scaling) only kick the can down the curb. Further, network outages can completely cut access to the database. Thus, it is imperative to maintain the database across different machines (horizontal scaling) to ensure smooth access even when the network is unreliable. Maintaining copies of the database, or replication, results in a distributed database. Replication comes with its own set of questions: how do we ensure that these copies are up-to-date with each other at all times? Measuring Database performance Before we continue, it’s worth discussing how we can compare databases based on their desirable properties. From distributed systems theory, the CAP theorem defines the landscape of database performance. It states that any distributed data store cannot provide all of the following three properties: Consistency, Availability, and Partition Tolerance. Consistency (C) is the guarantee that queries to the distributed database always happen in a certain order. Thus, all records in all copies of the database must be composed by the same set of operations. For example, all copies of a MySQL distributed database must have the same tables at all times, and must have built these tables using the same INSERT, UPDATE, etc. statements. Availability (A) entails that any request to any node in the database always returns some kind of non-error response. This includes scenarios where nodes in the distributed system have failures, where the response will have an indefinite delay. Partition-tolerance (P) refers to the system’s ability to operate after losing/delaying an arbitrary number of messages between nodes. Specifically, when the network is segmented into groups of nodes, or partitioned, some partitions will not be able to communicate reliably to others. Thus, a distributed system cannot have CAP as there can be two partitions, G1 and G2, of nodes within the system that don’t communicate with each other, and a write query to G1 followed immediately by a read request to G2 will have inconsistent values as G2 will not show the write query to G1. image This problem is avoided in an AC system as requests can always be communicated between nodes. It is also avoided in AP systems as the inconsistency from the above situation is not required to be solved, and in CP systems the database queries will not succeed as availability is not required. In a real-world network, hardware failures are largely unpredictable, and thus partitioning is typically assumed. As a result, AP and CP systems are preferred over AC systems. It is important to note that CAP is a rather simplistic quantification of database performance, but serves as a good starting point to understand other metrics. Leader-Follower Paradigm When dealing with a cluster of nodes, our goal is to ensure that the data is properly replicated on multiple nodes. Therefore if a node goes down, then the data does not “disappear”.The cluster can simply look into another node for the relevant data. The most common and traditional approach to replication within distributed systems is a Leader-Follower approach. Each node within a cluster is separated into two groups: leaders, which accept write requests and send the data to the other group, and followers which can only accept read requests. The general idea is for the leaders to accept writes, and then send the data change contained with the write to the followers to copy from. The read requests are mainly handled by the followers to take the load off the leader, while its resources are focused on accepting the writes. image I. Single leader System This type of system designates one instance in the cluster to be the leader, and the rest of the replicas to be followers. This paradigm is often useful when writes need to follow a sequential order. The single leader can sequentially process every write that occurs. For example, three writes may be requested to multiply, then divide then subtract from a value. Since these writes are not commutative, it is essential that they are dealt with sequentially. A write occurs in the following manner: the leader receives the write request, then writes the data to its own storage and then sends the same write to the followers to make the data change as well. A write can be accepted under two different conditions. Firstly, the replication from leader to follower could be done asynchronously; the write could be accepted after the leader writes the data to its own storage and replicates on at least one other node. Or the write can be accepted synchronously; after the leader is sure that each of the followers have also replicated the data change to their own store. This subtle differentiation has implications on consistency and availability. Asynchronous vs Synchronous Replication Synchronous replication works as the following. A write is sent to the database. The leader handles the write by updating its own data store. Then, the leaders send the data to the followers for them to replicate. Each follower sends back a status stating that data was successfully replicated to their data store. The leader waits for each follower to send a confirmation response. After ensuring every follower has successfully replicated the data, the leader informs the client that the write has been successful. This ensures that the data is consistent across all nodes. Consequently, when a read is requested, the cluster sends the request to any data store. However, this level of consistency comes with a trade-off; a more latent and therefore less available system. Take for instance, a cluster with twenty instances that observe a single leader, synchronous approach, i.e one instance is designated as the leader, and the other nineteen are designated as the followers. A write request is sent to the leader. The leader processes it and then waits for the status of the replication from each of its followers. Perhaps, multiple followers are currently running slow and take longer than usual to replicate the data and notify the leader of the data change. The leader is held into a predicament where it cannot handle new writes since it is waiting for the data to replicate amongst the rest of the followers. Thus the system can no longer guarantee availability. Asynchronous Replication in contrast lets the leader deal with write requests without having to worry about every follower replicating the data change. The leader accepts writes, makes the change to its own data store and then sends a notification to the followers to replicate the data from the leader. The leader then begins accepting new write requests. However, since the data is not guaranteed to be consistent this can have unpleasant consequences. Imagine a person wants to update the profile picture on their social media account. This change can be considered a “write request”. I.e. UPDATE Profile SET Picture = ‘newpic.jpeg’ WHERE ProfileID = 1; The follower would accept the write and handles the write asynchronously. The person then checks their profile to ensure the change was made. This request would send a “read” request to the cluster. The person is delighted to see that their photo change has occurred. This would be a case of the read request being routed to a follower that has successfully replicated the data change from the leader. Then the person notifies a friend to comment on the picture. The friend goes on the person’s profile but notices no profile picture change has occurred. This would be a case of the read request being routed to a follower that has not successfully replicated the data change from the leader. An approach that combines the two is the minimum insync replica approach. The data system sets a number, n, denoting the minimum insync replicas. When the leader deals with a write, it only waits until n followers have successfully replicated the data. This way the leader is not bogged down by waiting for all followers to replicate the data and the system is thus less likely to become unavailable due to waiting for write requests to succeed. Node Outages In a distributed system, one pitfall that may occur is node churn, which is when nodes shut down arbitrarily. Therefore if a node goes down our system can still be running, accepting write and read requests. To deal with node outages, it is important to know what type of node went down. If a follower node goes down but comes back up, then how would it catch up to all the data changes that occurred? If a leader goes down, how does the system continue accepting writes? Follower Outage Each follower keeps a log of the data changes it has received from the leader. Therefore, the follower knows the last transaction that occurred before it went down. When the follower comes back up, it can request from other nodes all the data changes that occurred while it was down. This effectively catches the follower up, and makes it consistent with every other node in the network. Leader Outage More complexity is introduced when a leader goes down. To ensure the system continues accepting writes, one of the followers in the system needs to be elected a leader. Then every follower needs to be aware of this change so that it could replicate data from the newly elected leader. This process is called failover. Failover can be done manually by the administrator or it could be done automatically. The automatic steps are generally the following. Determine the leader has failed. Leaders can fail in multiple ways. Network outages, power failures, crashes and more. There is no foolproof way of checking if the leader has gone down but this is usually done by timeouts. If the node doesn’t respond for a certain duration, i.e. 20 seconds then it could be understood that the leader has failed. The next step is to choose a new leader. The best candidate is usually the follower with the most up to date data. This ensures that data loss is minimal. Getting the nodes to elect a leader is considered a consensus problem, which will be explored in further blogs. The final step is to reconfigure the system so that clients now send data to the new leader. Also, it has to ensure that if the older leader comes back on, then it knows it is no longer the leader, but instead a follower. There are many common pitfalls with failover. For example, in asynchronous replication, an older leader may come back up. However, it may have data that was written to it that other followers have. How should the system deal with this data? Another problem is understanding when the leader has gone down. What is the correct timeout length? A longer timeout would make the recovery of the system longer if the leader has failed. A shorter timeout may make for unnecessary failover. For example, if the system is experiencing network delays because of a spike of activity, then making the system do a failover process will lead to more load on the system thus further slowing it down. Replication Lag Single leader architectures are well suited for application workloads that consist of mostly reads and a small percentage of writes. Think of a simple blog website. Most of the traffic is users reading the blogs, while the writing occurs only by authors, or users that want to comment. One way to scale this website is to add more nodes, i.e. more followers to handle the read-only requests. However, as previously noted as more followers are added to the clusters, so does the time to replicate data amongst them. Therefore an asynchronous approach or minimum insync replicas becomes more realistic. As stated before, asynchronous approaches are not consistent. Instead, they are considered “eventually” consistent. That eventually all replicas will catch up and copy the data. However, there is no guarantee on how long this will take. It could be a few seconds to even a few hours. This is known as replication lag; the time it takes for the data to be replicated amongst all replicas. There are many unexpected problems that could happen with replication lag. We’ll explore one in detail and solution to it, namely monotonic reads. Monotonic Reads: Imagine the following situation. You read a new blog that was just posted. You go to comment on the page, but your wifi goes down. After reconnecting to the wifi after a few moments, you again navigate to the blog to make a comment. However, there is no blog to comment on. It seems that the website is going back in time. What has happened is that initially, you have read from a replica that is up to date. But then the subsequent read request is sent to a node that is not up to date. Monotonic reads is a guarantee that the above would not occur. Namely, a user will not see an older replica, after it has seen data from a more up to date replica. One way of achieving this is for the user to read from the same replica. This could be done by choosing a replica based on some function of the user’s identity. However, if the chosen replica goes down then the user’s request will have to be routed to another replica. II. Multi Leader System The largest downside of a single leader approach is that only one instance is a leader. For example, a network interruption between the leader and a client could effectively stop that client from writing to the database. An extension of the single leader approach is to have multiple leaders within the system. Imagine you have multiple data centers separated geographically. In the single leader approach, one replica has to be the leader of all these data centers. In the multi leader approach, you can have one leader for each data center. The other nodes are followers to the leader within their data center. The leaders in turn follow other leaders. image A multi leader approach makes more sense for the following reasons. Firstly, such a system performs better. In the single leader system, every follower will have to route to the single leader that may be outside it’s data center. In the multi leader approach, the followers only have to look at the leader in their local data center. Secondly, in the case of a data center going out; a single leader approach would have to elect a new leader, if that data center contained the leader. However, in the multi leader approach, writes can be accepted since there are multiple leaders. The most glaring problem with the multi leader approach is that some data may be concurrently changed within two different data centers. Write Conflicts in Multi Leader Systems Consider the scenario where one write request could set a variable to a certain value, while another write request simultaneously sets the same variable to a different value. In a single leader database, the second write will either block or wait for the first write to be done. It could also abort the second transaction and ask the user to resolve it. In the multi leader setup, both writes are successful and it is later asynchronously detected at some later point in time. There is no guarantee when this will be detected thus making it too late to ask the user to resolve the conflict. The way to solve this problem is through conflict resolution. In a multi leader configuration there is no defined ordering of writes, so it’s not clear what the final value should be. Thus, every replication scheme should ensure that the data is eventually the same on all replicas. One popular way to achieve this is through Last Write Wins(LWW). Let’s briefly go over the general structure of it: Give each write a Unique ID (using a function that ensures each ID is unique). Pick the write with the highest UID as the winner, and throw away the other ones. However, this approach is prone to data loss, since the other writes are discarded.Moreover there may be writes that are discarded that are not concurrent. Resolving a write conflict in a multi leader approach can be solved and minimized using various strategies. There are many strategies to look over, that we won’t go into full detail. Some notable conflict resolution strategies include Last Write Wins (LWW), Operational Transformation, Mergeable persistent data structures, and Conflict Free Replicated Data Types. Conclusion Thank you for making it this far. In this article, we introduced the need for a distributed system as a means for safely storing and retrieving data. We covered the leader-follower paradigm as a way of nodes interacting and exchanging data amongst each other, under the assumptions that the network could be unreliable and node outages may occur. The Leader-follower paradigm helps the database reach a consistent state amongst nodes, however there is a cost since the system must always agree upon a leader, and that a leader must always be present. This leads our system to be less available in certain cases. New paradigms have emerged to address these issues. Namely, the leaderless paradigm where no node is a leader and thus no need for expensive operations such as Failover in the case of a leader outage. In the next article, we will introduce the leaderless paradigm and elaborate on how leaderless systems are more available and how they keep the system consistent. Citations - Kleppmann, M. (2019). Designing data-intensive applications the big ideas behind reliable, scalable, and maintainable systems. Beijing: O’Reilly. https://martin.kleppmann.com/2015/05/11/please-stop-calling-databases-cp-or-ap.html - Seth Gilbert and Nancy Lynch. 2002. Brewer’s conjecture and the feasibility of consistent, available, partition-tolerant web services. SIGACT News 33, 2 (June 2002), 51–59. DOI:https://doi.org/10.1145/564585.564601