Restate 1.2 - Distributed Deployments - Community Meeting February 2025

So thanks Giselle, my name is Ahmed as she introduced me. I lead the system design for distributed ReState and pretty much all my experience has been around distributed systems and storage. So let me, I guess, share my screen. All right, can you see my screen? Okay. Yes. All right, so I'm going to quickly go through the high availability features that we are shipping with ReState 1.2. Some of these might be a little bit more advanced, but I'm going to try and explain at least what you're going to see or what you're going to observe as a user. But also happy to go into the details if anyone is interested. So for the inpatient, I wanted to find a way to explain how you interact with ReState in the simplest way possible. And I came up with this. So this is the first run. Let's see if it works. So if you interact with Restate, how you use it? Well, it's a set of steps. Basically, first thing is you develop a service. So you write RPC-like service. You use Restate SDK. And you think about a piece of logic. You design it similar to an RPC. And then you deploy that service somewhere. This could be on your own server, on AWS Lambda, or Cloudflare Worker, or anything. basically a whole lot of options for you to deploy your service. Then once you deploy that service, you get a restate server anywhere and you register that service to it. So now you have teaching one restate server somewhere that, hey, this is my service. This is where it's deployed. Once that service is registered, now you need to call your service. So when you call your service, you call your service through restate. So you don't call your deployment directly. you call ReState Ingress. So ReState Server gives you an Ingress port. This is an HTTP port. There are, of course, there's also the Kafka Ingress, but then this is just to simplify things. So now you're talking to your service through ReState. So ReState in that sense acts like your API gateway. Then once you call your service through ReState, ReState Server will call the deployed service on your behalf. And it will maintain the bi-directional communication between the server and the service. This means that even if your service is deployed behind the balancer, or if you have multiple instances deployed in different maybe cloud providers, ReState will kind of handle that for you. So if there's any state, any resumption of invocations, everything basically is handled through the journal and the communication between ReState server and the SDK. This all happens behind the scenes. So yeah, so that's kind of the workflow once you have that profit. So when we say RestateService, we mean your code running on your own deployment, deployed somewhere. When we say RestateServer, this is RestateServer, our server, that has an ingress port and has also other ports for its communication. And when you invoke a service, you invoke service by talking to RestateServer, not to your service. You talk to RestateServer, RestateServer proxies that. and adds its magic. Now, this all works well, looks fantastic. But what if I don't want to have a single server, like a restate server? Similar to how my servers can be deployed behind LoopBalancer, and I have many of these, I want to have a bunch of restate servers for a few reasons, scalability, high availability. I want to be able to update restates while everything's still running, and I don't want to have any downtime, et cetera. So that kind of transition from single server to a cluster is what Restate 1.2 is about. So let's go through that. So your interactions with Restate goes through two things. So I'm just going to visualize here the two key things we have explored. So one is, OK, I have my server deployed. This is the one on the right. and you're registering the service that you're teaching the Restate server where that service is deployed is by talking to the admin port or admin API. It's the same port, by the way, that runs the user interface, the UI that Nick has been kindly presenting. Once the service is registered with Restate, it gets stored in some sort of internal storage to kind of remember all the details about that service. So what sort of functions, all the schema. Right? So, well, this is now stored locally here. Okay. Then when you invoke the service, this is another view. So when you invoke the service, just to get to visualize, for example, if you do curl, so my restate here, slash my service, slash my key, slash do work, which is an example function in your service, you're calling my restate. My restate here is the, let's say the domain name or the host name of the restate server that you're communicating with, not their service. deployment. So the server deployment is completely hidden from like reset users. And even if you update it, like deploy the different version, that URL that you're talking to doesn't change. But what happens when you have multiple reset servers? So one of the first challenges that we're discussing is, okay, if you have a cluster of reset servers, we would want to have, well, for every reset server, there is ingress port, right? It has its own domain name or a host name. And we would want this to be kind of completely transparent to you. So you don't know which is which. Inside ReState, you can talk to any ingress server and send your requests to anyone. And then whether there is internally like leader election or any failover or anything, it should be completely transparent to you. So now in the next ReState 1.2, once you run a cluster, you're going to have multiple ports, multiple... uh ingress endpoints to talk to this depends on which server you need to communicate with and it's up to you really to uh load balance across them so you can put a little balance in front uh to kind of distribute the loop but regardless of how you communicate uh or or which ingress you talk to the requests you need to eventually go to the right server serving the specific partition for example for a particular particular object that you're communicating with or a particular service and some of that will be uh have been explained uh by tilt in the previous presentation so i think that's recorded on youtube if you want to kind of take a look at it so it's a lot of details about uh partitions uh and i'm building basically on top of it here all right uh now clustering support means that the usual kind of system on chip design that we have um first it needs to continue uh to be system on chip so we we have this philosophy of single binary um no dependency and we wanted to have this way even with clustering so in that world you would be running multiple of those systems chips and or binaries on different servers and they they need to kind of create a mesh internally and communicate uh with restate this is uh prior to 1.2 but i think it becomes more prominent in 1.2 is that you now can run restate the same binary but you specify which roles the binary has And this gives you ability to disaggregate, for example, portions of the system that need more compute versus storage. So the same binary, but then you can pause with roles. And by default, it runs all roles, which is just to make it simpler for everybody. But you can decide that you can specialize certain parts of your deployment to perform certain roles. This also newly in 1.2 includes ingress. So you can just need a clip. feature flag and then you can like a configuration option and then you can even say that only these nodes can serve as ingress and no one else will but by default everything is enabled all right now that you have this stuff so kind of going through these as quickly as i can so what have we shipped first thing is we shipped a new um replication system for in the like one of the kind of critical system critical parts of the system which is the log the log is the durable interconnect that connects the different machines um in a durable fashion so a kind of storage layer uh this is the direct immediate persistence layer for all the state machine changes for all partitions and this is based on flexible quorum uh consensus so uh it gives you what while we are shipping the or exposing through our APIs and our configuration, the simplest version of this, the underlying technology allows for way, way more powerful replication options. So you can have fine-grained control over your write availability versus read availability. And even GOA replication, which hopefully is going to be one of the features we're going to be releasing in the near future. And the second thing is partition replication. So partition is the materialized state. And now we also want to replicate this. And the partition replication is not quorum-based, which means that it's a leader-follower style. So you have a leader partition that just reads the log and keeps the state in memory and also stores it on disk. And it keeps kind of replaying. you can have multiple followers for the partition. The followers for a partition basically follow the log. But if you have followers that are down for the partition, or if you configure your partition replication to be one, this is independent from the log replication. So the log replication can be 3x, 4x, whatever. And every time you create a partition or you start a node, if it doesn't have the partition information, now it needs to reconstruct the partition state from somewhere. And if it already had, follower partition running that means that it has the state warm and this makes kind of failover much much quicker but if it doesn't have it it can fetch well also one of the newly added features here a snapshot that's automatically and periodically uploaded to an object s3 or compatible object stores or any any um like min io or s3 or anything and it will so it will fetch that replay the bits in the log that uh from the snapshot to the tail and now you have a partition So this means that failure in one of the partition followers transitionally doesn't impact availability by any means. You don't need a quorum for partition replication, but you need quorum for log replication, if that makes sense. The number three is the, like also ran out of the oven, is a native replicated metadata store. well in order to continue the replication story so we need to have a replicated log and replicated partitions so for availability and for a quick fillover but also we need to have uh one of the critical pieces of the replicated metadata store and this is raft based it's a very critical piece but it's also the piece that has the least load so we the majority of data goes through partitions and the log but raft only keeps very very small bits of information that are critical but also rarely updated or not updated as often. It's not in the critical part of the system. So we only update them usually on failover or when machines come up or go down, et cetera. But on a normal operation, basically, it's just almost idle. And this is critical, right? This is really important because we didn't want to have huge dependency on a super high-performant metadata store, whether it's internal or external. So this is also brand new. completely integrated with the system. And at most, it keeps a couple of megabytes of data, maybe less than 10 megabytes at worst case. And the last thing is, well, now we have the set of machines in the cluster running, and we have the replication, but requests that come through the network, so from Ingress, the one that I described, now need to be routed internally to partition leaders and so on and so forth, and handle things like failover, what if leader. switched from one node to the other and ingress that received the original request is a completely different node so now you need to have like an internal routing network and uh this is this like one of the principles here is that we want to give this as transparent as possible so this is also part of that what is shipped and if you put all these things together um we get um like completely highly available um replicated restate um that you have also fine green control over each one of its layers So going into details of each one, well, quickly as well. So the distributed log portion of this, which is the one that basically handled all immediate durability, we call this bifrost. So if you see bifrost in documentation or configuration file, I think it's the only piece that we sometimes call it by name, like give it a code name. Everything else is just plain old normal words, but bifrost is the kind of thing. past interconnect that we have it's really critical for the latency so sometimes you you might see this and um it's flexible a core on base as i said and it has a few nice optimizations so kind of it's it's aware of uh locality uh it's um it has like it's optimal really tuned for restate workload and this is what makes it very special in comparison to any log system So this is kind of tail optimized because most reads happen at tail. So this is the last few records. So they are cached in memory. They are co-located on the machines that really need it. So the sequencer and so on and so forth. And also one of the two key features here is like zero copy reconfiguration. So if you are changing your replication factors, if you want to go from like 2x to 5x, for example, it doesn't mean that I want to copy all the history to 5x. I only want to kind of reconfigure the new data that comes in this makes it very quick to reconfigure and you don't have to pay the cost for all your historical data in fact your historical data doesn't matter that much because your everything gets materialized very quickly to the partition store and at that point once purchase store is our the partition database is snapshot it to object store you you can basically oh this automatically has a kind of green light to trim the log up to this point to save disk space. And the replicated metadata store, I mentioned already a lot of these details. It's pretty small. And we only store a handful of keys there. Its API internally is very simple. And again, it needs to keep a few megabytes at worst case. It's kind of low write, mostly read metadata style things. So we store membership information, so which nodes are members in the cluster, which roles they play in some leadership information about partitions and the information about the log system. So it's a segmented charted log, all of that metadata about these segments and replication factor and all that stuff should basically get stored in there. This is a replicate partition. I think I skipped the slide very quickly. I didn't notice. So that's the metadata stuff. all the values are versioned monotonically as well. This is also a part of the fundamentals of the restate design. The replicative partitions is what I mentioned earlier. So leader follower style replication, it's non-quorum replication, and it gives you the ability to control warm follower field over and cool snapshot of a strap. So you can save compute and storage by running leaders only. You don't need followers. And at failover, you pay the cost by failover time, where we need to pitch a snapshot, rehydrate the partition somewhere else, which takes a little bit longer but saves you compute and storage. Or you can have worm followers, like actual followers always running, tailing the log, and materializing the state as you go. And then on failover, we just very quickly switch leaders from here to here. And it's up to you to figure it this way. And yeah, snapshotting also allows you, snapshotting as we said, like it goes to object store and transparent routing is the last bit. So you call any ingress, your traffic gets routed correctly. If there's a failover that's happening during your ingress call, you're gonna get a retriable error. So your HTTP client can retry the request and it will be rerouted. This is the same for also everything. So restates. CTL, the new command line tool, or any restates, like a command line tool, the normal restates CLI, and restates CTL, CLI, and the UI, or any SQL queries, for example. You talk to any node, and it figures things out for you. You don't have to kind of point your client to a specific node. Anyone will work. So that's kind of the idea we would like to, and I think this helps explain some of the demos that Taylor's going to show. So Restate cluster is going to look something like this. Your services are kind of in the middle. Now we have this triangle where Restate servers are interconnected. For every server, there is, of course, a bunch of partitions running, in this case, highlighted as like P1, P2, just an example. By default, we run 24 partitions, but you can configure this as you like. At the moment, we don't support changing the number of partitions, so choose this number carefully in the beginning, but hopefully also in the future we will support expanding and shrinking the number of partitions, so dynamic re-partitioning. The yellow means it's a leader partition and leader elections internally and dramatically. The gray here is like a warm follower. Your services are connected, or any reset server can connect to your service. And the curl commands here denote how you reach out to your server. So you talk through any of these ingress ports that we open to run the same. curl or whatever uh http request but they basically just need to change which reset server you're pointing to uh this is going to be useful i think when we do the demo because if a server for example goes down so if one of these three servers goes down uh you need to talk to different ingress uh restate might still be running but you're talking to an ingress that's actually down so it's very useful for example to run a low balancer on top of recent ingresses we wanted to do this with uh just one last point and we're going to go to questions one last thing is uh this is we Why don't you do this with self-contained design? So this is something that we don't want to compromise on. And it's just a reminder. All right, let's go to questions or points. Yeah, there's a question by Pavel. What is the difference between the Restate and Restate CTL CLIs? Right, so Restate is Restate CLI. The normal CLI we have been shipping is the CLI for Restate users. So this is anybody who's kind of... building on ReState or a developer or any user, ReState CTL is more towards operators of ReState clusters. So if you are an admin of ReState cluster, it gives you a little bit more of internal details, like where partitions are running, who's the leader right now. It's kind of a low level, more low level component. We might in the future merge some of the parts together into ReState. We are still, like ReState CTL is more, It's in infancy, right? And we are still kind of shaping, so don't rely so much on its interfaces or its command structure. It's less stable than the, not stable as in, like it doesn't work, stable as in its shape, like the command structure shape is not as stable as the restate CLI, normal restate CLI. So normally, if you do not use restate clusters, you will not need restate CTL at all. Right. A few other questions. Kevin asks, can you expand on the long-term plans for geo-aware partitioning and data placements? I can share some plans. I'm not sure if we are at the position in which we can share everything, but I can give you some things that we have been basically building for. And also, I want to kind of put this as a disclaimer. Some of those maybe advanced features for geo-replication might be more tuned towards enterprise customers or enterprise users as a nature. So I'm not sure if we have exactly defined where the lines are for, for example, if there will be some of those features that will be behind like a paid subscription license or anything. That's just, I kind of want to put this as a disclaimer on top of this. That's still not clear. But I can tell you what we can do, like what we're building for. So the geo-aware replication gives us the ability to, well, first we have two layers or three levels of replication, right? Three different systems internally we need to replicate. Metadata store. So I'm going to put this on the side. i'm not going to talk a lot about it but i'm going to talk about the partition replication and log replication uh they serve different purposes for both um well more importantly first the log because this is where the quorum is uh we the gra replication means that you're gonna have to say well i want my data to open my log to be replicated across let's say two availability zones or three evaluators or two regions and you're going to have flexibility it's going to it's a tree structure hierarchy where you say it's like a three level so there's region um zone and node so if you say node 5 or node 2 which is uh what's currently exposed at the api is it means that i want to replicate across two nodes right that's it any nodes any two nodes but if you say zone 2 uh it would mean that i i want to replicate and guarantee that if one zone goes down that all my data will be available in the log uh or it can be region two uh zone three all right which means that i have my data replicated across two regions uh but internally across three zones so it's like three zones distributed across two regions so three is for availability uh across zones but but two for regions is so that you can tolerate a regional failure that is on the log side very similarly on the partition side is gonna work quite the same except that you don't have quorum requirement so it will be mostly placement i want to place my partitions on three regions And hopefully we're also going to have the policies for affinity. So you can say, well, I prefer this region or I prefer, like most of my users, for example, are in this availability zone. So ReState will try and optimize placement by default in that region if you have capacity. Otherwise, it's going to fall back into other regions or other zones if we lose our getability there. Or maybe Till has already answered, no? I think a few other questions. Why do we need two logs, the replicated log Bifrost and the replicated partition with the embedded database? Why not keep everything in one log using consensus and use that as the ground truth? It is one log. Are there two logs part of each and every partition? No, it's just one log. So there's one log. When I say replicated partition... it actually means that just the partition is running twice. There is one liter, the other one is forward. Both are reading from the same log. So both are constructing the materialized from the same log. So that's the replication of the partition. It's just to get your materialized state warm and ready to failover. At the moment, we have one log per partition, but this is not a strict rule. The number of partitions to the number of logs is something that we can change in the future. It's an internal detail. And that log, by the way, whether the number of logs and how they correspond to the number of partitions from a user perspective, it's like a one log. It's just a sharded log. It's a kind of one logical log, basically. Let us know if that answers your question, Janaki. A question by RJ. Is there any recommendation on how to pick the partition number? I think already mentioned some of this, right? It's our line of thinking. So a partition, let's think what a partition means, what its impact. So a partition can be thought of like two things. One is a compute unit. So at the moment we run like a thread for partition on a machine. So if you, while this might not be, might continue in the future, it's just an implementation detail at the moment, but It's a good way to think about it as if it's like a core, like a CPU core. So you try to run maybe twice the number of cores in terms of partitions locally, or two or three, just because partitions are mostly IOO bound. So you can actually run a little bit more than your CPU cores. Of course, it depends on what your partitions are doing. But in most cases, they will be. So twice the number of cores is kind of a good start. This is why our default is like 24, which tries to kind of go into like a normal, maybe modern laptops, double number of cores or so. But that's one thing. The other thing is it's a storage unit. It's the unit of failover. So if you failover or need to pick up a snapshot from object store, it determines how big, how much data you need to fetch. So more partitions means smaller units of storage, which means quicker feelover, but also means more fragmentation and more overhead. So it's a trade-off, right? Just a good starting point is to think about it as the number, of course. So if you have a cluster, think about the number of partitions every machine needs to run. It depends on your replication property, how it's set, and just double that number as a guided factor. hopefully in the future it's going to be less of an issue once we have all dynamic partitioning so it's purchased are going to be automatically split and unsplit based on the load reactively so i have a follow-up question on that like if we have more partition you mentioned like it would the snapshot would be smaller would that mean that we would write more snapshot to s3 and basically have more api calls to s3 Yeah, that's also true at the moment, yes. So more partitions means more databases, more smaller databases that you need to do. But in grand scheme of things are the number of calls and the number of API calls that you need to do. It's relatively small in grand scheme of things. These are in the tens or maybe hundreds. But even if you run like a thousand partitions, let's say every hour you're going to be snapshotting, like making a thousand calls to S3, which is... I think there was a question about observability. How observability is exposed when things are... Giselle, please let me know if we went way beyond time. I'm not sure. Maybe we should first do the demos and then answer the remaining of the questions? Yes, I think the demo will be exciting. So I'm going to hand this off to Till. just just to make sure that if anybody needs to leave at seven that they still see a bit of the demo and then we can take it as long as we need to after the demos to go through the questions yes yes all right let's hand this off to the king of demos take it away thanks amit um yeah hi everyone until um i spend most of my time um helping amit building the the distributed runtime that i want to show you now And as a first demo, I want to show you how Reset's high availability works. And for that, let me actually share my screen. And very concretely, here you go. Let us first take a look at a Docker Compose setup. So what I want to show you is a three-node Reset deployment. And then I want to show you what you can do with the three-node setup in terms of crashing nodes and make sure that that reset still can make progress. So the first thing here, that's like a Docker Compose setup. It defines for all nodes some common... configuration, which is the cluster name so that the nodes know that they all join the same cluster. All three nodes that we start run all of the roles that Ahmed mentioned that a reset server binary can run. So it runs the admin role, which exposes the admin API, and runs the cluster controller, the worker role running the patching processor. the log server role running the replicated loglet store and the metadata servers responsible for storing the metadata one important piece that we need to configure when when deploying a multi-node reset cluster is that we that only a single node can actually provision the cluster being like the seat of the cluster to which other nodes join and then expand the cluster. That's why by default, we set the auto provision flag to false and can either provision the cluster via the reset CTL CLI or by configuring a single node to be the seat of our cluster. Then in order to run in a cluster setup, we set the bifrost provider like the lock provider to replicate it before in we said 1.1 it was local we say that we want to replicate our data to at least two nodes so that we can tolerate a single node failure that's also why why like i i will spin up three nodes so that we can tolerate an arbitrary node failure and we also configure this metadata server type to be replicated so that we can also tolerate metadata server crashes. If you remember maybe the demo from a couple of months ago, there it was, we still had the metadata server not being replicated and needed to run on a node that we couldn't crash, but this is no different. And it needed to have like also single admin. Now this also is gone, like you can run admin everywhere. Yeah, exactly. And in order for the nodes, or the metadata clients, to talk to, to find each other, to update the nodes configuration and fetch the information about the other nodes, we need to tell which nodes are part of the initial set of nodes where you can find the current nodes configuration and the membership information of who are your peers in the cluster. Apart from that, A bit further down, we just define three nodes that run actually a bit more than 1.20. It's the upcoming 1.21 that I want to show you. They just define their node name and node ID, the addresses, and we say that the first node is actually the one that is allowed to provision the cluster, like to seed it, from which then the cluster grows and becomes the multi-node cluster. and the other two nodes aren't allowed to do that. And that's it, basically. So quite, quite simple. As you see, no external dependencies, just the reset binary that we are running. All right, now let's switch over to my shell, and let's actually spin it up. Docker Compose up. That should do something. um we see that we are starting a couple of reset server nodes uh running 1.2.1 and they seem to start their partitions okay so far so good um by the way if you have any any questions any point then then just um we can use the restate ctl to show uh that they are yeah right yes it's here right exactly spoke too soon so now i want to show you um the the new reset ctl cli which we can use to actually now inspect our our running cluster so and to do this um it's it's running on also on the on the nodes and that's why i can just do like docker compose exit on node one and call reset ctl status which prints us under the status of our current cluster which is that we have three nodes we see their names as we have defined them in the in the docker compose file they all are part of the metadata cluster so we have a three node metadata cluster and we see that that some of the nodes do run a partition leader like like node one runs partition leader leaders for for three partitions and node two for a single one and node three was the unlucky one it got only followers assigned um yep and that's basically we have a running cluster now if you're interested in a bit more information you can also pass the extra flag which gives you a bit more information about what's the the current version of the the metadata that the class is operating on and more details into how the log is currently configured. So here right now we have a replicated log, which is crucial for the high availability, which replicates to two nodes so that we can tolerate a single node failure. And here we see the progress or the status of the different partition processors, like what is the last applied LSN. And as you will see in a second, when I create invocations, and you see that the cluster is making progress and at the very bottom you see some more detailed information about the metadata and store all right next thing is i would configure or like register the um the the service that i have already started um this service actually let's do let's do no here let's do it here it's a bit bigger i think This service is a very simple service, it's a counter service where we can count values like just incrementing a value that we keep in restate state and clear this counter. It's a virtual object, so we can choose a key and count for every key separately. So let's register it. All right, if we now do... list our services we see that we have our counter registered perfect now let us put some load on it and for this we have developed a small little program that that sends invocation requests to the three available ingresses so on the very left window we are sending the invocation request for counting counter one to node one as it's written here let's do the same for for node two so we're sending an invocation request to node two as well on a different key of the counter and line three for node three okay so we have three uh processes now running that invoke services on every request on every ingress of the the cluster let's actually take a look at what the partitions are doing as we are invoking services here the interesting bit is now that we we see that the that the applied lsn column is increasing. So as we are sending invocations to the cluster, they get stored in Bifrost. And from there, they are being read and processed by the partition processors. And as they process more and more invocations, they make progress or the applied LSN will increase. So we also see that there's one partition that doesn't get any invocations that you can see from ResetCTL. Because we only probe three different keys. Now, let's see how this thing behaves if we kill node one. We see that node one is actually the leader for three partitions. One, two, and three. So, Docker Compose kill node one. Okay, so the first thing is... um that you see here that our request to node 1 fail which is also makes sense because we killed we just killed the process um that also runs the ingress um so the the request from from this this client fail however what this also shows is that the the client client program that i'm running here um if it fails to send send a request to to uh ingress it will try um one of the other ingresses in this case it sends the request to node 2. And we do see that actually the counter is correct. So the requests just get routed a bit differently, but will be processed by the respective partition that still has all the data because it was previously written to at least two nodes. So we can tolerate a single node failure. All right, if you know. maybe take a look at the status again let's see what what happens oh no true is that yes that's true so if we run reset ctl status again we do see now that node one is indeed offline so that one is dead um and the the leaders the partition leaders and also the um the the sequencer the the replicated locked sequences from this node where move to the remaining nodes node 2 and node 3 okay let's actually restart node 1 this should hopefully fix the problem as you see perfect now let's see actually what happens if i kill more than one node so let us take first first a look at the status again just to see everything is running and we do see that um that that our leaders are currently node two and node three so let us kill node two first compose kill node two okay um this should um actually let me check what um just happened oh yeah okay so after killing node 2 so the cluster just needed to um well basically reconfigure itself um like like moving all the leader tasks that we're running on on node 2 to different nodes and this means um reconfiguring the the lock like sealing the existing lock uh for for for those logs that were where the sequencer was running on node 2 and moving the the partition processor leader to a different node which at times can take a few seconds to to um be be be done okay now let us kill a third node and let's see what happens if we kill a third node the system should um um should come to a halt because we have lost the the right and and white availability um there's only a single node left to which we can write but we set um to to the system that we require at least two replicas of our data to in order to tolerate a single node period right so that's why um while the system as you see it here um stops it just tries to to append but it can't find the two nodes to to append the data to um this is um like this is like the the intended behavior in this case because otherwise if we would just allow like a single node uh writes then then if this node dies the data might be lost Sebastian asks a question, sorry I wasn't really looking at the demo at the time, but Sebastian asks, why did we have to restart node one? Shouldn't it figure out that node three is back on its own? Did we restart node one to let it know that node three is back? I think in the scenario I killed node one and then I just brought it back to bring the cluster up again to up to three nodes so that I could kill another node again. yeah yeah so the command was just a docker compose restart node one just to get the node back up not really a command to restate to tell it oh yeah sorry i should have made clear that this was a docker compose instruction to restart the the node um but let's actually try it out now again with um node 2. now i'm instructing docker um to to restart me node 2 and hopefully but what should happen is that the system would recover from the um from having lost right availability and and continue making progress as it left off so let's see whether this happens first thing is it needs to to figure that out by yeah we see here on the left it's continuing also the the increments are continuous so we didn't lose any any rights perfect the only thing is that node 3 is still dead but our third client here will reroute the request to node 1. we can actually check our cluster status again so here we see that now we have actually two nodes alive n1 and n2 we have our four logs with a replication of two which is fine and we see that some of the nodes are now in catch-up mode because they haven't applied the the latest or aren't at the tail of the lock so they need you to catch up with what has happened but it should by now it should have happened and all of them let's see should be active perfect and that's Just for the sake of completeness, let's restart node 3 as well, so that we bring the cluster to full health again. Here now the node 3 that we have just restarted needs to catch up because it was offline for a bit of time where it missed some of the new entries, but after some time they have fully caught up and the cluster is healthy again. All right. So much for the first part of the demo for high availability. Are there any questions? The log, if you look at the logs, it will tell you a lot about what's going on behind the scenes. So if you're interested in the details and the integrals, feel free to read the log output. Also, I think this is the frequency at which we are sending these requests. It's very gentle, just to let you see. the activity but it doesn't mean that this is the fastest rate that you can send ingress requests it's just because otherwise if we if we let this counter go much faster it's going to be very difficult to observe what's going on yeah for for those of you who are interested in some some performance numbers what the system can achieve under load they are included in the release blog post i believe And if I remember correctly for a single step workflow running on three nodes, it was the ballpark of being able to process something like 23,000 requests per second. Yeah. If you do this actually on your laptop, so if you have a MacBook, for example, on a MacBook, you're going to do out-of-the-box 24,000 requests per second. with no configuration. There are a few questions, so we can, until you have switched reviews, that I can maybe echo or answer. So one is about whether we are testing with Jepsen, right? This is Sebastian, are you going to do in DST or you'll be doing that? And I'll actually answer this. This is just for audio. No DST at the moment, but we, We have actually continuous Jepsen tests that run nightly and on-demand. Right now, actually, we are enabling this for every PR. So we do this for network partitioning testing and various failures. There's also another test suite that we have, which is the verification test. This is actually a pretty nice suite of tests that maybe one day we're going to have a lot of details written about it. So it's a way to generate very complex test programs that test various parts of it's deterministically generated. So it generates a huge set of operations and goes through all partitions, runs it for a few hours, and then collects and runs failure scenarios at the same time. So we crash machines, delete disks, and we want to make sure eventually that these programs finish. correctly so anything that adds numbers or deletes or subtract numbers eventually the answer is correct so we want to kind of deliver that consistency and correctness is not violated under any failure scenario um holding the system is an acceptable outcome if you lost so many nodes uh but what's not acceptable is uh losing data or uh incorrect results so state machine doing something incorrect but that is that is there's a question for you tail and what happens if you lost the disk so we said is able to detect a disk loss and and put in that case the the node into a disk loss or data loss state from which you as an operator need to to um and well you you need to check whether this actually um really like a loss of data that you need to act upon or whether this can also happen in case of a misconfiguration of your system if you for example start a new node with a name that has already been used in your cluster in which case you can actually get out of the data loss state by using the restatectl Yeah, so this is something that the system can detect and warn you about. Yeah. There's also ability, so with the snapshot restore, so if you lost a disk and you feel specifically a partition database, the node will fetch the snapshot from the snapshot store and replay automatically. So this is without intervention, so it is configured correctly. So is that all the demos or do you have other demos? I have one more demo. um for those of you who are interested in actually upgrading uh reset 1.1 to 1.2 uh and from a single node to a multi-node uh cluster if you if you are interested um so that's what i want to show you now um let me just share my screen again okay so to show you this i have prepared this different docker-compose setup and the the first docker-compose that i want to run runs just reset 1.16 okay then i will register a service and and invoke some some services to generate some data that the system needs to store and then i want to actually upgrade this this node to reset 1.2 and afterwards i want to add more nodes to to grow from a single node to a multi-node setup so that they actually have um for tolerance and high availability so let's try this out so i have here my docker compose the version 1.1 um all right here i have um yeah we said server 1.16 is running so fast so good let us register um the service again the um the counter service let us generate some some load so we just add a few or count a few times we see um here new invocations are being processed perfect so we are at count value 12. now let us actually um stop the the 1.16 reset server and um let's upgrade it but before let's let me show um the difference between the the what you need to to change in terms of configuration to make this happen um so this is the was the old um configuration of our 1.16 um reset server and now this is the one for 1.2 the main change like is that we switch our metadata server type from local to replicated that's it everything else is something that you then when the cluster is running you need to reconfigure to go from a single node to multi node and for this to work also what i've done is configured a snapshot repository to which the the node can can create a snapshot of the partition processor state from which the other nodes can resume that is needed because with the with the local loglet with which the 1.16 server is running there is no support for remote reads it only supports like in-process reads so there can't be a different process reading from from this loglet that's why we do need to create a snapshot um and trim the lock once we have reconfigured the lock but i will show you that apart from that it's again the very simple docker compose setup just here that this time i'm also spawning a min io docker container which which i use for the the snapshot storage that's it okay let us try this out now so Let us docker-compose start node 1 and the min.io container. That's all I want to do. Okay. Okay. So the first thing you will see is that we are trying to migrate the local metadata to the replicated metadata, because now we are running with a replicated metadata server. and apparently this was successful yay and the system seems also to be running okay um let us actually check this out with our status command and here we see um now we have a single node cluster running because it's now we said 1.2 we can use resetctl to query its its status so far so good one quick check let's look at the detailed information we do see that we still have the local log provider configured let us check whether we have actually recovered the metadata properly by listing our services so we have the counter cool and now let us check the count of our actually adding or counting counting up by one by by calling our service again remember we left off at 12 so when i run this command it should be return 13. and it does so um we have successfully migrated we have successfully migrated 1.1 reset server to a 1.2 reset server without doing losing any any data and only by switching the configuration option metadata service type from local to replicated that's all okay cool um now let's let's try actually to to add some some notes to the cluster um to be able to do this we first need to reconfigure it so we need to reconfigure the lock from local to replicate it so that other nodes can actually do remote reads from it. So we do this by using... So, I'm going to explain this. One quick thing I wanted to give, like those nerdy distributed system folks that are tuning in. So one of the things that we have designed here is that we wanted to have the ability to transition consensus algorithms or replication protocols or even... storage systems, for example, like this one, live. So you wouldn't need to wipe the data. It was part of the promise of v1 release is that we're going to maintain compatibility. And even transitioning from single node to cluster, you wouldn't need to go and wipe everything. And so even that we are technically changing the log implementation, right now what Taylor is going to do is changing the log implementation from a local log to replicated log. And same for the metadata. uh it happens transparently uh and without um well without downtime so like you wouldn't even need to restart or stop the server to do this so this is a demo kind of that demonstrates not just like a migration path but that it happens live yeah exactly so let us move now to a switch the the the locklet um to the replicated locklet um um we do this here um we say Yes, we set actually, so we chose the log provider to be replicated and the log replication to be one because we only have one node currently in our cluster. So it wouldn't make sense. Or if we configured anything higher, the cluster wouldn't be able to choose the right node sets for the replicated loglet and basically lose white availability until you have changed this configuration again. so remember maybe for one second that the last applied lsn for partition zero was 91 and for the other two for the other partitions it was two and let's then quickly remember also this view and how it looks uh in a second once we have reconfigured so i i have instructed the cluster now to to use the replicated lock that from now on and on on a change of a log provider from local to replicated this will be eagerly applied to all the available locks so let's check this out now we have actually bumped the locks version to v3 from v2 and we see that we now have a replicated lock let that starts from lsn98 for partition 0 for 1 to 3 from lsn3 so basically all new appends will now um or lock appends will now happen in in the replicated locklet um at the moment it's still not like we still have only a replication of one because there's only one note but we will change this in a second and next thing to to be able to to add new nodes to your cluster is i need to create a snapshot because if i would now start a new node it would because there's no snapshot yet it would try to read from the beginning of time so from from lsn1 but lsn1 is is stored in the local locklet which is not reachable from from the outside so that's why i need to create a snapshot trim the lock and then i can start new notes let's try this out now i can use for this the reset ctl command again snapshots create i do this we see we have created some some snapshots great it also tells us the the contained lsn the min lsn that the snapshot contains now i can trim the log so i i do this for all the four available logs um oh yeah because makes sense um i had a um off by one error um okay um so now it should be trimmed and now we should be able to to start new nodes let's do this here okay compose up node two so first i only add a single node let's see what happens keep fingers crossed that things are working so we started a second node first it um it so it found a partition snapshot it first tries to um okay here is an invalid key range for Let me actually quickly check whether I screwed up with the number of... Is this the same cluster that you ran this morning? I mean, it could be that you're... I actually changed recently the number of partitions from six to four. Oh, okay. We don't support repartitioning yet. Yeah, yeah. Maybe I should wipe very quickly the disk. um that's the fun of uh doing live demos yes um how do we uh also to give everybody kind of a pointer do we have um a tutorial on the wiki uh or on the on the sorry on the website uh on house migrate right yes i can post a link for me share a link yeah also it's a it's a great idea if you're gonna do this is to backup the system, so copy all the restate directory and do this on a just in case, right? It's very easy to clone restate and run this. There was a question also I can answer until you get things set up again. There was a question from Apostolos. I might have butchered the name, sorry. Is the cluster setup production ready? configured or it requires fine-tuning skills. The defaults that we ship are quite reasonable, so you should be able to run this on most hardware. I think there are just a couple of tunables that you may mainly need to consider yourself with if you don't want to go into the integrated details. The number of partitions is one, which you have seen actually here in this demo as well. So the default number is 24. You might want to increase this or decrease this depending on the setup. The other one is the memory you allocate for restate, which I, if I remember correctly, the default was six gigabytes. Depends on how much memory do you have in the system and how much you want to give restate, you also can configure this. More memory gives us more caching space for various bits of the system. However, the default should be fairly sufficient for a small deployment. If you want to do this on a larger scale or run many nodes or heavily utilized cluster, I think you'll need to tune things a little bit more. And there's a few tunables. What I would suggest is to start small, experiment, run real estate, figure things out, and also report back if there's anything that's a little bit unexpected or some of the things that might not be tunable that needs to be added to the configuration file that we can also edit for you so hopefully this answers your question if if you're still around i think also we are a bit over time so we can actually skip this one if uh if all possible doing this on docker might be very tricky um yeah i i might probably have screwed up the configuration uh it was a very um short notice um preparation for for this part of the demo um but i'm very happy to um to deliver this uh the next um meeting if you like

Restate 1.2 introduces highly-available distributed deployments. Ahmed Farghal and Till Rohrmann explain the internals and give demos on high availability and upgrading Restate clusters. 

Restate 1.2 announcement blog post: https://restate.dev/blog/announcing-restate-1.2/
Restate architecture deep dive: https://restate.dev/blog/the-anatomy-of-a-durable-execution-stack-from-first-principles/