Artboard

3e:null 3f:["$","title",null,{"children":"A Causally Consistent Redis Backend for a Realtime Database | fstfwd | Observable"}] 45:T517,${await FileAttachment( "43537da5-2c25-4b47-9fd0-11b0a117482e_tlark_httpss.mj.runFpimq4__dropped_capital_letter_F_on_medieval_parchment_in_red_and_gold_with_hand_drawn_.jpg" ).image({ style: "width: 4em; float:left; padding-right: 0.1em; border-radius: 0.3em" })}irebase's databases may be easy-to-use and intuitive, but they actually solve some of the trickiest problems in distributed databases. **Firebase is often misclassified as an eventually consistent database**, but if that were the full truth, you would hear developers complain of data loss footguns and logically "impossible" security holes. The reason why Firebase is intuitive to use, even for inexperienced programmers, is because a phenomenal amount of engineering resources were dedicated in ensuring the database achieved the best possible consistency guarantees you could hope for in a distributed setting *i.e.* **causal consistency**. Causal consistency provides the bedrock on which other advanced features, such as offline persistency, optimistic updates, can be built upon. In this notebook we advance our [hackable Firebase-compatible server](https://observablehq.com/@tomlarkworthy/firebase-server-prototype-1) by developing a Redis backend suitable for use in a serverless architecture.46:T583,## Causal Consistency The technical definition of *causal consistency* is that all clients *observe* operations in an agreed *causal* order. Causal ordering is defined by the *happens-before* relation. The weedle room in that definition is that non-causally related operations can be reordered relative to one another. ([Time, Clocks, and the Ordering of Events in a Distributed System](https://lamport.azurewebsites.net/pubs/time-clocks.pdf) by _Leslie Lamport_ is a classic paper on the subject) As an individual client will generally only observe a small portion of the global dataset, they will only assert constraints on the ordering of the things they *read* and *write*. I tend to think of client's continuously making statements like: *"I saw the data at location was X so I performed mutation Y"* is how these causal relation are realized in practice. In the case of *Firebase* where all clients connect to a central server, _causal consistency_ implies that **clients always see _other_ client's operations in the order they occurred**. There may be significant delays to observing the operations of an individual client — it could be offline — but when it does sync, those operations are observed in order. Operations made by different clients, on the other hand, may be observed in any order relative to one another. ${await FileAttachment("cc@1.svg").image({ style: "max-width: 400px;" })}47:Tb59,The stronger distributed consistency model, *sequential* consistency, requires a global ordering of operations and doesn't permit progress during network partitions. So sequential consistency is a non-starter for a distributed setting with ephemeral participants like web page visitors. Going the other way, *eventual* consistency, implies nothing other than operations eventually become visible, but says nothing about ordering. Eventual consistency can be the cause of hard to replicate race conditions, and can leave you second guessing root causes of bugs. Causal consistency, on the other hand, is the optimal middle ground, which seems to match programmer's expectations of semantics. The ergonomics of causal consistency is one of the reasons why Firebase was such a hit. So suffice it to say that **causal consistency is what you want in a distributed database, and the key is preserving the order of operations** 48:T477,In my imagination of a better Realtime Database server, I think, **I want to use an ordinary programming language for authorization rules** and, I want to **query over different data-sources through federation**. Within auth checks, I want to be able to **call 3rd party APIs**, and write inefficient code if I want, **synchronously**. What I don't want to do, is to conform to some *deliberately constraining bespoke language* which is what Firebase asks me to do. Firebase, and similar products, don't allow general programming languages in the critical path because authorization is part of the database engine where latency impacts throughput. To enable end user customization, we need to rethink the architecture. Luckily new options have appeared since Firebase was developed. Modern infrastructure, like serverless, offers new opportunities for horizontal scaling. So let's try to execute our authorization rules and other services in horizontally scaled stateless workers. With this architecture, slow authorization or federation code would cost slightly more to run, but it would be isolated from the primary database performance. 49:T60f,## Redis as the Shared State In this new architecture, we use shared state that all the stateless workers will need to access every request. We chose Redis, as it is offered everywhere, it is known for low latency, and because we can also use it as the primary storage too. In this architecture, the stateless workers terminate public client connections and execute the database protocol by moving state around a Redis backend. Redis has some fantastic features for ensuring *at-least-once delivery*, *exactly-one* processing and therefore _causal consistency_. Firstly, queues, hashmaps and stream are all first class citizens in Redis with their own specialised instructions for manipulation. Secondly, sequences of Redis commands can be applied atomically in a [transaction](https://redis.io/docs/manual/transactions/). For example, we can pop an instruction from an inbox, apply the effects, and write notifications to other inboxes in a single atomic operation. This is an amazing because if we can apply client operations, including all the event deliveries to all other clients, in atomic operations, we can meet the requirements for _causal consistency_. The main requirement for _causal consistency_ is that other clients observe a client's operations in the order they were raised. So, if we pop operations from a clients inbox in order, and ensure they are applied atomically, then the effects on other client's outbox observations are also in causal order, and we are free to execute instructions from different clients in parallel. 4a:T444,suite.test( "Client operation queue: enqueue_operation, next_operation, ack_operation", async () => { const client = await createClient(redisConfig, "operation_queue_client"); init_client(client); enqueue_operation(client, { request_id: "1", payload: "init" }); // If we write twice with the same request_id it is deduplicated enqueue_operation(client, { request_id: "1", payload: "dedupe_test" }); enqueue_operation(client, { request_id: "2", payload: "second" }); const next1 = next_operation(client); const next2 = next_operation(client); ack_operation(client, "1"); const next3 = next_operation(client); ack_operation(client, "2"); const next4 = next_operation(client); expect(await next1).toEqual({ request_id: "1", payload: "init" }); expect(await next2).toEqual({ request_id: "1", payload: "init" }); expect(await next3).toEqual({ request_id: "2", payload: "second" }); expect(await next4).toEqual(null); } )4b:T663,suite.test( "smoke test process_operation with LISTEN/PUT/GET/UNLISTEN", async () => { const client = await createClient( redisConfig, "process_operation_smoke_test" ); clear_data_listeners(client, "foo"); init_client(client); enqueue_operation(client, { request_id: "1", action: "PUT", key: "foo", value: "baz" }); enqueue_operation(client, { request_id: "2", action: "LISTEN", key: "foo" }); enqueue_operation(client, { request_id: "3", action: "PUT", key: "foo", value: "bar" }); enqueue_operation(client, { request_id: "4", action: "GET", key: "foo" }); enqueue_operation(client, { request_id: "5", action: "UNLISTEN", key: "foo" }); while ((await process_operation(client)) !== "NOOP") {} const history = []; var next = await next_notify(client); while (next) { const [id, reply] = next; history.push(reply); ack_notify(client, id); next = await next_notify(client); } expect(history).toEqual([ { request_id: "1", status: "ok" }, { action: "DATA", data: "baz", key: "foo" }, { request_id: "2", status: "ok" }, { action: "DATA", data: "bar", key: "foo" }, { request_id: "3", status: "ok" }, { data: "bar", request_id: "4", status: "ok" }, { request_id: "5", status: "ok" } ]); } )4c:T60b,## Conclusion and Next Steps So this notebook implements a Redis backend for a *causally consistent* backend for a Realtime Database. We should perform some fuzz testing before considering the semantics finished, but I would prefer to integrate this into a working 3rd party Firebase-compatible server first, so we can fuzz using the vanilla Firebase clients. Note the real Firebase Realtime Database server has more semantics than we have implemented. Its LISTENs can query and filter, its values can contain JSON documents, so this is far from a feature complete server backend. However, under the literate programming paradigm, I prefer each notebook to concentrate on a single topic. This notebook is about *causal consistency*. We will add query semantics in dedicated notebook later. They can all be found in the [Firebase collection](https://observablehq.com/collection/@tomlarkworthy/firebase). I spoke with [*Andrew Lee*](https://startupandrew.com/about/), the former CTO of Firebase about this post, and he said that the original Firebase Realtime Database *was* prototyped on Redis. This explains why Firebase's wire protocol has an identical PONG for a PING command like Redis does! Anyway, Firebase did not go further with Redis because they needed disk persistence to be economical in the multi-tenancy use case (this was 2014, Redis Flash options were thinner then). So going back to Redis perhaps represents a full circle for Firebase. Anyway, I hope you have enjoyed this writeup! Please ❤️ at the top and spread the word!4d:[{"id":"dcaef342095f3be1dab70b47943b99879e58dc529328c05be704e141703e928b09b1f476c1b44c2a077a4ff1e0e96a0fee4eb411c6103c6ee75fb76213f27dac","href":"https://static.observableusercontent.com/files/dcaef342095f3be1dab70b47943b99879e58dc529328c05be704e141703e928b09b1f476c1b44c2a077a4ff1e0e96a0fee4eb411c6103c6ee75fb76213f27dac","name":"cc@1.svg","type":"image/svg+xml","lastModified":1702584591448.805,"size":213583},{"id":"983bc063173c7a92038c7ae5da914679660dbecbd957d5c1d3e64e75b5ec6db5f7b9e3005e42fcf6f6afbc9f21e51b7cdc8aebc55ecbe1db6205f89fda2db484","href":"https://static.observableusercontent.com/files/983bc063173c7a92038c7ae5da914679660dbecbd957d5c1d3e64e75b5ec6db5f7b9e3005e42fcf6f6afbc9f21e51b7cdc8aebc55ecbe1db6205f89fda2db484","name":"a.svg","type":"image/svg+xml","lastModified":1702584591448.805,"size":285605},{"id":"bf4ad5b16896b3d6d2944e02f7f23bd6f67a36b1e085c9d255bf7f757fc4a4cf51dbe05a70d6aaed0974bd8849e203bc67a2aef4a257a984579bca40daca3a65","href":"https://static.observableusercontent.com/files/bf4ad5b16896b3d6d2944e02f7f23bd6f67a36b1e085c9d255bf7f757fc4a4cf51dbe05a70d6aaed0974bd8849e203bc67a2aef4a257a984579bca40daca3a65","name":"Q.SVG","type":"image/svg+xml","lastModified":1702584591448.805,"size":46747},{"id":"7216a502ce536fbb7cd66f865a878b13f2aa40a1f96c342c91c4ad8a9679d89b00dfd2c1a31780d6079892048e4eba4ab4c5455a27d513b010354e313ff71e1f","href":"https://static.observableusercontent.com/files/7216a502ce536fbb7cd66f865a878b13f2aa40a1f96c342c91c4ad8a9679d89b00dfd2c1a31780d6079892048e4eba4ab4c5455a27d513b010354e313ff71e1f","name":"l.svg","type":"image/svg+xml","lastModified":1702584591448.805,"size":3900},{"id":"ab6ca1e70f72cc5b97db5df4a5200c8bae82aa9bc4204d7d9da3b203ac5614e19728806d385512e248634e107e88841392abce0423619e8c86ed2edadda035d7","href":"https://static.observableusercontent.com/files/ab6ca1e70f72cc5b97db5df4a5200c8bae82aa9bc4204d7d9da3b203ac5614e19728806d385512e248634e107e88841392abce0423619e8c86ed2edadda035d7","name":"t.svg","type":"image/svg+xml","lastModified":1702584591448.805,"size":10729},{"id":"e94b1c86d07f5dc09f6f01934a2b7df26282abde98eff458009a61cf02fff6beff4e9efc7ac2901ba0de61e5ff9f41fa3cf47b307a627950f5965ab2404246e7","href":"https://static.observableusercontent.com/files/e94b1c86d07f5dc09f6f01934a2b7df26282abde98eff458009a61cf02fff6beff4e9efc7ac2901ba0de61e5ff9f41fa3cf47b307a627950f5965ab2404246e7","name":"43537da5-2c25-4b47-9fd0-11b0a117482e_tlark_httpss.mj.runFpimq4__dropped_capital_letter_F_on_medieval_parchment_in_red_and_gold_with_hand_drawn_.jpg","type":"image/jpeg","lastModified":1702584591448.805,"size":133007},{"id":"8d7e13a86bdbcc9b9252b183def7273cc17709d4c42f3488e79860677165e5e5f87598acf032d4a03b0bc36a39d15c56bc65d4d906340c2bec8e6635be2e5fe7","href":"https://static.observableusercontent.com/files/8d7e13a86bdbcc9b9252b183def7273cc17709d4c42f3488e79860677165e5e5f87598acf032d4a03b0bc36a39d15c56bc65d4d906340c2bec8e6635be2e5fe7","name":"firebase_secret@4.jpg","type":"image/jpeg","lastModified":1702584591448.805,"size":46521}] 43:{"id":"b64e995364939f37","slug":null,"title":"A Causally Consistent Redis Backend for a Realtime Database","thumbnail":"29a8f68d90364b73c2c9cca5cd75a7387ddb73103d54e4668a5b9d0028382bf5","access_level":"public","create_time":"$D2023-12-14T20:09:51.448Z","update_time":"$D2023-12-14T20:09:51.448Z","view_time":null,"first_public_time":"$D2023-12-14T20:09:51.448Z","has_unpublished_edits":null,"fork":{"id":"de798a70f81e9e09","slug":"redis-backend-1","title":"A Causally Consistent Redis Backend for a Realtime Database","owner":{"id":"7db5ed2b0697d645","login":"tomlarkworthy","name":"Tom Larkworthy"}},"stars":0,"starred":false,"owner":{"type":"team","id":"8017900b13067003","login":"fstfwd","name":"","avatar_url":"https://avatars-next.observableusercontent.com/303dc4f18a5797db38267406fe4a7645181b0cd35eacd60259337786987fc4d9","tier":"starter_2024","bio":"","following":false},"version":4435,"publish_version":null,"role":null,"authored":false,"authors":[{"type":"individual","id":"36112d77d168de8f","login":"fstfwd","name":"","avatar_url":"https://avatars-next.observableusercontent.com/303dc4f18a5797db38267406fe4a7645181b0cd35eacd60259337786987fc4d9","tier":null,"bio":"","following":false}],"members":[{"type":"individual","id":"36112d77d168de8f","login":"fstfwd","name":"","avatar_url":"https://avatars-next.observableusercontent.com/303dc4f18a5797db38267406fe4a7645181b0cd35eacd60259337786987fc4d9","tier":null,"role":"editor"}],"comments":[],"custom_thumbnail":false,"trashed":false,"lapsed":null,"subscribed":null,"forks":0,"collection":null,"type":"notebook","body":{"titling":"auto","title":"A Causally Consistent Redis Backend for a Realtime Database","cells":[{"id":4010,"value":"$${await FileAttachment(\"firebase_secret@4.jpg\").image({\n style: `max-width: ${Math.min(640, width)}px`\n})}\n# A Causally Consistent Redis Backend for a Realtime Database","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4022,"value":"$45","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4173,"value":"toc(\"h2,h3\")","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":4196,"value":"$46","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1787,"value":"$47","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":3999,"value":"## A Literate, End User Programmable Implementation on Redis","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4060,"value":"This notebook is a prototype of a Redis based backend intended for the next iteration of a [**hackable 3rd party Firebase-compatible database server**](https://observablehq.com/@tomlarkworthy/firebase-server-prototype-1). This notebook is live code! the code is executed in *your* browser. Its will eventually become part of a **transparent cloud** where all services are auditable, extendable, self-hostable and [remixable](https://nickarner.com/notes/remixable-software-guest-post-may-31-2022/). This notebook includes integration tests that exercise the broad features. The implementation connects to an external Redis server which you can verify in your browser's development tools. If you fork this notebook, you can change the code, the commands will hit the same external Redis server unless you also change the [redisConfig](https://observablehq.com/@tomlarkworthy/redis-backend-1#redisConfig) cell below.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":8,"value":"redisConfig = ({\n socket: {\n host: \"redis.webcode.run\",\n port: 443,\n tls: true\n }\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":4209,"value":"If you have any questions, you can ask them inline with a [comment](https://observablehq.com/@observablehq/comments).","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2500,"value":"viewof suite = createSuite({ name: \"Reactive Integration Tests\" })","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1776,"value":"## Why we are building this: end user programmability","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1783,"value":"*(BTW, this comes from painful personal realization from reflecting on my work on the [Firebase Rules](https://firebase.blog/posts/2018/01/introducing-query-based-security-rules), [Blaze Compiler](https://github.com/googlearchive/blaze_compiler) and the [Common Expression Language](https://github.com/google/cel-spec))*","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1527,"value":"$48","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4218,"value":"a = FileAttachment(\"a.svg\").image({ style: \"max-width: 600px\" })","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":4216,"value":"\nHorizontal compute, like serverless, add additional design constraints though. Workers are ephemeral and stateless. But this statelessness clashes with stateful clients like Firebase that require executing a carefully ordered stateful wire protocol to meet their consistency guarantees. We will resolve this conundrum by externalizing all state into a stateful backend that the workers connect to, and for this prototype, we will try Redis.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2122,"value":"$49","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2231,"value":"### Connecting to Redis: *createClient*","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2207,"value":"We will dedicate Redis client for each Firebase client. We will identify a client by its **client_id**, which will map to Firebase's connection id ([source](https://github.com/firebase/firebase-js-sdk/blob/8bece487710aa6315c7dd98bcb086cd18fc9a943/packages/database/src/realtime/Connection.ts#L98)). ","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1453,"value":"createClient = async (redisConfig, client_id) => ({\n redis: await createRedisClient(redisConfig),\n client_id\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":10,"value":"import { createClient as createRedisClient } from \"@tomlarkworthy/redis\"","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2476,"value":"So that we can run functionality in *this* notebook, let's create an example client now!","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":461,"value":"exampleClient = {\n restartClients; // Depend on restartClients cell so we can cycle this client\n return createClient(redisConfig, \"exampleClient\");\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2484,"value":"As Redis connections are stateful, if we hit a bug it is useful to cycle them","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1047,"value":"viewof restartClients = Inputs.button(\"restart clients\")","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2488,"value":"We can test the client is working by sending a PING command. Lets put this in a unit test so we can confirm a working connection to a Redis server. ","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4434,"value":"exampleClient.redis.sendCommand([\"PING\"])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2511,"value":"suite.test(\"Example client responds to PING with PONG\", async () => {\n expect(await exampleClient.redis.sendCommand([\"PING\"])).toBe(\"PONG\");\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2228,"value":"### Initializing the state for a fresh client: *init_client*\n\nCentral to the ordering guarantees is that client operations are processed in order, implying a queue for each client. \n\n${await FileAttachment(\"Q.SVG\").image({style: \"max-width: 640px\"})}\n\nWhen an SDK connects for the first time to the server, we need to setup two queues streams for the inbound (operations) and outbound (notifications) directions. We use Redis [streams](https://redis.io/docs/manual/data-types/streams/) to implement the queues, as they have features for ensuring *at-least-once* delivery which are beyond the basic [lists](https://redis.io/docs/manual/data-types/#lists).","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":481,"value":"init_client = ({ redis, client_id }) => {\n clear_operations({ redis, client_id });\n clear_notifications({ redis, client_id });\n set_head_operation_id({ redis, client_id }, \"0\");\n return set_head_notify_id({ redis, client_id }, \"0-0\");\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2411,"value":"#### Key helpers\n\nAll data-structures in Redis are stored under keys. We use helper functions for generating key names for per client data-structure.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":455,"value":"client_operation_queue_key = (client_id) => `c-${client_id}-actions`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":942,"value":"client_notify_queue_key = (client_id) => `c-${client_id}-replies`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2415,"value":"#### Clearing streams","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2383,"value":"Clearing streams is as simple as issuing the [DEL](https://redis.io/commands/del/) to the key holding the stream.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":890,"value":"clear_operations = ({ redis, client_id } = {}) =>\n redis.sendCommand([\"DEL\", client_operation_queue_key(client_id)])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":965,"value":"clear_notifications = ({ redis, client_id } = {}) =>\n redis.sendCommand([\"DEL\", client_notify_queue_key(client_id)])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2431,"value":"#### Manipulating the head of queue pointer\n\nFor *ordered at-least-once* delivery, a queue should shift up **after** the action has been applied. This implies we need keys to remember where we are in the queue, which will be incremented as we do the work. Redis streams already have entry IDs, so we will reusing that.\n","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2443,"value":"First we define helper functions for the keys","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":499,"value":"client_operation_head_id_key = (client_id) => `c-${client_id}-actions-head`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":945,"value":"client_notify_head_id_key = (client_id) => `c-${client_id}-replies-head`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2448,"value":"Then we define helpers for setting them. The [SET](https://redis.io/commands/set/) command updates the key with the provided value.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":521,"value":"set_head_operation_id = ({ redis, client_id } = {}, value) =>\n redis.sendCommand([\"SET\", client_operation_head_id_key(client_id), value])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":955,"value":"set_head_notify_id = ({ redis, client_id } = {}, value) =>\n redis.sendCommand([\"SET\", client_notify_head_id_key(client_id), value])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2459,"value":"Of course, later we will need the [GET](https://redis.io/commands/get/) too.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":503,"value":"get_head_operation_id = ({ redis, client_id } = {}) =>\n redis.sendCommand([\"GET\", client_operation_head_id_key(client_id)])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":949,"value":"get_head_notify_id = ({ redis, client_id } = {}) =>\n redis.sendCommand([\"GET\", client_notify_head_id_key(client_id)])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2522,"value":"So after we initialize a client, we expect their head pointers to be 0-0. Let's confirm that with an integration test:","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2524,"value":"suite.test(\"init_client resets queues and heads\", async () => {\n const client = await createClient(redisConfig, \"init_client\");\n enqueue_operation(client, {}); // Put some data in the queue\n\n init_client(client); // reset the client\n\n expect(await get_head_operation_id(client)).toBe(\"0\");\n expect(await get_head_notify_id(client)).toBe(\"0-0\");\n expect(await next_operation(client)).toBe(null);\n expect(await next_notify(client)).toBe(null);\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2550,"value":"## Initializing a new long poll session: *createLongpollSession*\n\nInterfacing to Firebase's long polling transport requires some additional metadata to be tracked across requests. Its not hugely relevant to *causal consistency* but just something we need. We use Redis' [HSET](https://redis.io/commands/hset/) and [HMGET](https://redis.io/commands/hmget/) to associate a dictionary of values for each client. So far we are only using it to store the long poll session password.\n\nThe session password is a security measure that prevents other agents manipulating a running long poll session ([source](https://github.com/firebase/firebase-js-sdk/blob/master/packages/database/src/realtime/BrowserPollConnection.ts#L54)).","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1574,"value":"client_longpoll_session_key = (client_id) => `c-${client_id}-lp`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1567,"value":"createLongpollSession = async (client, { password } = {}) => {\n init_client(client);\n client.redis.sendCommand([\n \"HSET\",\n client_longpoll_session_key(client.client_id),\n \"password\",\n password\n ]);\n return {\n client,\n password\n };\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1587,"value":"retrieveLongpollSession = async (client) => {\n const response = await client.redis.sendCommand([\n \"HMGET\",\n client_longpoll_session_key(client.client_id),\n \"password\"\n ]);\n return {\n client,\n password: response[0]\n };\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2585,"value":"suite.test(\n \"createLongpollSession can set a password that is visible to retrieveLongpollSession\",\n async () => {\n const randomPassword = Math.random().toString();\n createLongpollSession(exampleClient, { password: randomPassword });\n const retrieveSession = await retrieveLongpollSession(exampleClient);\n expect(retrieveSession.password).toBe(randomPassword);\n }\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2625,"value":"#### Monotonic long poll response numbering with: *incrementLongpollResponseNum*\n\nThe Firebase database server numbers each server-to-client longpoll packet with a sequentially increasing response number. It is used by the client to derive the correct response ordering ([source](https://github.com/firebase/firebase-js-sdk/blob/c424340aaeedbf1772db06437e65ab252d7a90e5/packages/database/src/realtime/polling/PacketReceiver.ts#L20)). ","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1615,"value":"client_longpoll_response_num_key = (client_id) => `c-${client_id}-rp`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1623,"value":"incrementLongpollResponseNum = async ({ redis, client_id }) => {\n return (\n (await redis.sendCommand([\"INCR\", client_longpoll_response_num_key(client_id)])) -\n 1\n );\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2637,"value":"suite.test(\n \"incrementLongpollResponseNum increases by one each time\",\n async () => {\n const current = await incrementLongpollResponseNum(exampleClient);\n expect(await incrementLongpollResponseNum(exampleClient)).toBe(current + 1);\n }\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2647,"value":"## Database Semantics: Listenable Key-Value Store\n\nFirebase' data model boils down to a Key-Value store that clients can register for changes. So for a single data *location*, we need: 1, a place to store its *value*, and 2, a list of interested data *listeners*. \n\n${await FileAttachment(\"l.svg\").image({style: \"max-width: 640px\"})}\n\n*Firebase is a bit more complex, listeners can be queries over a collection, and the JSON structure cascades across data locations, but for this early prototype we are ignoring these details. *","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":811,"value":"data_key = (location) => `${location}-data`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":712,"value":"data_listeners_key = (location) => `${location}-listeners`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2676,"value":"The data holder can be accessed with Redis' scalar [GET](https://redis.io/commands/get/)/[SET](https://redis.io/commands/set/) operations. It's worth noting that we never *await* on these commands, as this will be pipelined in [transactions](https://redis.io/docs/manual/transactions/) later. I expect the data holder will be upgraded to a [RedisJSON ](https://redis.io/docs/stack/json/) representation in the future.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":821,"value":"set_data = ({ redis, client_id } = {}, location, value) =>\n redis.sendCommand([\"SET\", data_key(location), value])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":829,"value":"get_data = ({ redis, client_id } = {}, location) =>\n redis.sendCommand([\"GET\", data_key(location)])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2715,"value":"suite.test(\n \"Read our writes. get_data retreives data previously set by set_data\",\n async () => {\n const data = JSON.stringify(Math.random());\n set_data(exampleClient, \"read_our_writes_test\", data);\n const fetched = await get_data(exampleClient, \"read_our_writes_test\");\n expect(fetched).toBe(data);\n }\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2703,"value":"We store the registered *listeners* as a Redis [list](https://redis.io/docs/manual/data-types/#lists). This means we use [LPUSH](https://redis.io/commands/lpush/) (left push), [LREM](https://redis.io/commands/lrem/) (list remove) to add and remove to the list, and [LRANGE](https://redis.io/commands/lrange/) (list range query) to read the whole list.\n\nAgain, no *awaiting* in the code! these will functions will be pipelined and used in transactions.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":705,"value":"add_data_listener = ({ redis, client_id } = {}, location) =>\n redis.sendCommand([\"LPUSH\", data_listeners_key(location), client_id])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1336,"value":"remove_data_listener = ({ redis, client_id } = {}, location) =>\n redis.sendCommand([\"LREM\", data_listeners_key(location), \"1\", client_id])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":788,"value":"get_data_listeners = ({ redis, client_id } = {}, location) =>\n redis.sendCommand([\"LRANGE\", data_listeners_key(location), \"0\", \"-1\"])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1239,"value":"clear_data_listeners = ({ redis, client_id } = {}, location) =>\n redis.sendCommand([\"DEL\", data_listeners_key(location)])","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2728,"value":"suite.test(\n \"Listeners: add_data_listener, remove_data_listener, get_data_listeners, clear_data_listeners record client_id in a list\",\n async () => {\n const location = \"listeners-test\";\n clear_data_listeners(exampleClient, location);\n const afterClear = get_data_listeners(exampleClient, location);\n add_data_listener(exampleClient, location);\n const afterAdd = get_data_listeners(exampleClient, location);\n remove_data_listener(exampleClient, location);\n const afterRemove = get_data_listeners(exampleClient, location);\n\n expect(await afterClear).toEqual([]);\n expect(await afterAdd).toEqual([exampleClient.client_id]);\n expect(await afterRemove).toEqual([]);\n }\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2740,"value":"## Exactly-once operations\n\nIn normal operation, the Firebase clients request that the server to perform database operations, such as adding a data listener to a data location or setting the value of a data location. For the system to meet the _causal consistency_ goals, the order of a client's operations and the server's responses need to be preserved. To that end we use two queues per client. One queue for each direction of traffic.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2771,"value":"So when an operation comes in, we need to push that into a queue. However, consider what happens if we crash immediately after doing so? The client will not know whether we received the operation, and will resend. So, if we are not careful, we will push the operation onto the work queue a second time, and this will cause problems (operations are **NOT** idempotent).","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":2786,"value":"Firebase solves this by sending a *request id* ([source](https://github.com/firebase/firebase-js-sdk/blob/1a06d5d6064bd3b6b3672a352b0f128f86d6d89a/packages/database/src/core/PersistentConnection.ts#L181)) with each operation, so duplicates can be detected. *At-least-once* queues are neatly implemented in Redis using a stream and we use [XADD](https://redis.io/commands/xadd/) to enqueue. We can achieve *exactly-once* processing by using Firebase's integer *request_id* to reject duplicates, as the following implementation of *enqueue_operation* does.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":403,"value":"enqueue_operation = ({ redis, client_id } = {}, action) =>\n redis\n .sendCommand(\n [\n \"XADD\",\n client_operation_queue_key(client_id),\n `0-${action.request_id}`\n ].concat(Object.entries(action).flatMap((_) => _))\n )\n .catch((err) => {\n if (err.message.includes(\"ID specified in XADD is equal or smaller\"))\n console.log(`Non-monotonic request id ${action.request_id}`);\n else {\n throw err;\n }\n })","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2814,"value":"While *enqueue_operation* adds to the back of the queue, *next_operation* reads the front of the queue. Redis streams are similar to Kafka whereby readers maintain their read position in the stream. So first, *get_head_operation_id* is called to discover where the head is. This costs a round trip to Redis (note the *await*). We *could* eliminate the round trip by using [LUA scripting](https://redis.io/docs/manual/programmability/eval-intro/), but we will leave performance optimizations for another day. So instead we close our eyes and use the Redis stream instruction [XREAD](https://redis.io/commands/xread/) to read one item from the queue.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":444,"value":"next_operation = async ({ redis, client_id } = {}) => {\n const id = await get_head_operation_id({ redis, client_id });\n const response = await redis.sendCommand([\n \"XREAD\",\n \"COUNT\",\n \"1\",\n \"STREAMS\",\n client_operation_queue_key(client_id),\n `0-${id}`\n ]);\n if (response === \"QUEUED\")\n throw new Error(\"Can't use next_operation in transaction\");\n return (\n response &&\n response[0][1][0][1].reduce((obj, val, index, arr) => {\n if (index % 2 == 1) obj[arr[index - 1]] = arr[index];\n return obj;\n }, {})\n );\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2850,"value":"After calling *next_operation* a worker should actually try to apply the operation to the database. There can be two outcomes, the operation is applied successfully, in which case the *head_operation_id* should increment, or the operation fails, in which case we should retry. \n\nIncrementing the *head_operation_id* acknowledges the item in the operation queue was processed successfully, so we call this function *ack_operation* and it the stream id of the item we wish to move beyond. **We can achieve *exactly-once* processing if operation acknowledgement is in the same transaction as the operation's effect, thereby combining them atomically.** Though, we additionally need to exclude two processes from calling *ack_operation* twice for the same work with [WATCH](https://redis.io/commands/watch/) on the *head_operation_id*.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":495,"value":"ack_operation = ({ redis, client_id } = {}, stream_id) => {\n return set_head_operation_id({ redis, client_id }, stream_id);\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2879,"value":"$4a","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2905,"value":"## At-most-once notifications\n\nThe server responds to every incoming operation through a server-to-client notification stream. If any data is written to location with an active *listen*, the server will inform the client through this channel too. Because this channel contains operation replies AND data notifications, we cannot use the 1-at-a-time incremented *request_id* as the *stream_id*. This is kind of annoying as we can't form an at-least-once request pipeline, but it doesn't compromise consistency because if an notification goes missing the client can detect it and restart the session (how the detection works is transport dependant).\n\nLargely this code is the same as the queue for the *operation_queue*, the major difference is we let it auto-generate its own *stream_id* when calling [XADD](https://redis.io/commands/xadd/).\n","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":856,"value":"enqueue_notify = ({ redis, client_id } = {}, target_client_id, reply) =>\n redis.sendCommand(\n [\"XADD\", client_notify_queue_key(target_client_id), \"*\"].concat(\n Object.entries(reply).flatMap((_) => _)\n )\n )","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":3027,"value":"enqueue_notify(exampleClient, exampleClient.client_id, { payload: \"hi\" })","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":940,"value":"next_notify = async ({ redis, client_id } = {}) => {\n const id = await get_head_notify_id({ redis, client_id });\n const response = await redis.sendCommand([\n \"XREAD\",\n \"COUNT\",\n \"1\",\n \"STREAMS\",\n client_notify_queue_key(client_id),\n id\n ]);\n return (\n response && [\n response[0][1][0][0],\n response[0][1][0][1].reduce((obj, val, index, arr) => {\n if (index % 2 == 1) obj[arr[index - 1]] = arr[index];\n return obj;\n }, {})\n ]\n );\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":953,"value":"ack_notify = ({ redis, client_id } = {}, id) => {\n return set_head_notify_id({ redis, client_id }, id);\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":407,"value":"## process_operation\n\nProcessing an operation is the meat of the implementation. It is where we actually do database-like things like setting data values and registering listener. Of course, care must be taken to ensure we meet our causal consistency goals.\n\nSpecifically, *process_operation*, removes an item from operation queue, effects it and ACKS it **within a single transaction**. The transaction ensures the *operation* is applied atomically _i.e._ all or nothing. Some of the operation side-effects might include enqueueing *notifications* to other client's outbound message queues. Within the transaction, a [WATCH](https://redis.io/commands/watch/) on the *head_operation_id* ensures that **only one** worker can process this work task, leading to *exactly-once* processing. \n","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4291,"value":"

\n ${await FileAttachment(\"t.svg\").image({ style: \"max-width: 640px\" })}\n

Casual conistincy implies applying operations exactly-once and notifing observers through queues. This can be acheived in Redis within a transaction that does various datastructure manipulations.

\n","pinned":false,"hidden":false,"mode":"html","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"t","database":"$undefined"},{"id":2956,"value":"To help with observability during development, the functional interface is converted to dataflow using a [*flowQueue*](https://observablehq.com/@tomlarkworthy/flow-queue). This allows us to break the program flow across notebook cells so it's easier to diagnose how decisions are made. It also gives the autocomplete visibility into the data on the wire. So, the main *process_operation* function just forwards the function arguments to the beginning of corresponding flowQueue cell *process_operation_args*.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":416,"value":"process_operation = (client) => viewof process_operation_args.send(client)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":570,"value":"viewof process_operation_args = flowQueue({\n name: \"process_operation_args\",\n timeout_ms: 2000\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2972,"value":"The *process_operation_args* variable holds the last value seen: ","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":624,"value":"process_operation_args","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2982,"value":"First we fetch the next action after [WATCH](https://redis.io/commands/watch/)ing the *operation_head_id_key*. The WATCH prevents multiple workers doing the same operation.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":574,"value":"action = {\n const client = process_operation_args;\n client.redis.sendCommand([\n \"WATCH\",\n client_operation_head_id_key(client.client_id)\n ]);\n const actionRaw = await next_operation(process_operation_args);\n if (!actionRaw) {\n client.redis.sendCommand([\"UNWATCH\"]);\n viewof process_operation_args.respond(\"NOOP\");\n return invalidation;\n } else {\n return actionRaw;\n }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":3173,"value":"Some operations require additional data to be available before they can be fulfilled correctly. For a PUT operation, we need to know what the active listeners are. With the list of active listeners, a transaction can update the data location **and** notify the listeners through their queues in a single atomic update. To do this consistently, the prerequisite data must be not change during operation processing, so the gathered data is locked with [WATCH](https://redis.io/commands/watch/).","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":3164,"value":"prerequisites = {\n const client = process_operation_args;\n const operation = action.action;\n try {\n if (operation === \"PUT\") {\n client.redis.sendCommand([\"WATCH\", data_listeners_key(action.key)]);\n return {\n listeners: await get_data_listeners(client, action.key)\n };\n } else if (operation === \"GET\" || operation === \"LISTEN\") {\n client.redis.sendCommand([\"WATCH\", data_key(action.key)]);\n return {\n data: await get_data(client, action.key)\n };\n }\n } catch (err) {\n client.redis.sendCommand([\"UNWATCH\"]);\n console.error(err.message, action);\n viewof process_operation_args.reject(err);\n return err;\n }\n return {};\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":2996,"value":"After gathering the prerequisite data for the operation, we execute all the mutations within a transaction block [MULTI](https://redis.io/commands/multi/). Exactly what is mutated depends on the operation, which is handled by _operation_ specific flowQueues (explained later). ","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":605,"value":"process_operation_effect = {\n const client = process_operation_args;\n const operation = action.action;\n try {\n client.redis.sendCommand([\"MULTI\"]);\n const handler = {\n PUT: viewof run_put_operation_args,\n GET: viewof run_get_operation_args,\n LISTEN: viewof run_listen_operation_args,\n UNLISTEN: viewof run_unlisten_operation_args\n }[operation];\n\n if (handler === undefined)\n throw new Error(\"Unknown operation \" + operation);\n\n return await handler.send({\n ...prerequisites,\n client,\n action\n });\n } catch (err) {\n console.error(err.message, action);\n client.redis.sendCommand([\"DISCARD\"]);\n viewof process_operation_args.reject(err);\n return err;\n }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":3014,"value":"Finally, after the operation specific effects are queued, we acknowledge the operation to indicate it is processed, and execute the everything in an atomic transaction. Because *ack_operation* mutates the *client_operation_head_id_key* and we WATCHed it at the beginning, we guarantee that only one worker can do an operation. Because we also do the effect in the same transaction, we ensure we only *ack_operation* if the effect is applied. So we achieve *exactly-once* operation processing. \n\nThe response of the transaction is passed back to the enclosing [flowQueue](https://observablehq.com/@tomlarkworthy/flow-queue) using the *respond* function.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":660,"value":"ack_process_operation = {\n const client = process_operation_args;\n process_operation_effect;\n try {\n ack_operation(client, action.request_id);\n const response = client.redis.sendCommand([\"EXEC\"]);\n viewof process_operation_args.respond(response);\n return response;\n } catch (err) {\n await client.redis.sendCommand([\"DISCARD\"]);\n viewof process_operation_args.reject(err);\n return err;\n }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":976,"value":"$4b","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":3191,"value":"## Operation Effects","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":3966,"value":"operations = [\n put_operation_response,\n get_operation_response,\n listen_operation_response,\n unlisten_operation_response\n]","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":738,"value":"### run_put_operation\n\nA PUT operation overwrites a data location with a new value, and notifies all listeners of the change. Thus to apply the operation the liste of listeners at that location is required.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":743,"value":"viewof run_put_operation_args = flowQueue({\n name: \"run_put_operation_args\"\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":763,"value":"run_put_operation_args","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":778,"value":"run_put_operation_effect = {\n const client = run_put_operation_args.client;\n const action = run_put_operation_args.action;\n var result = set_data(client, action.key, action.value);\n run_put_operation_args.listeners.forEach((client_id) => {\n var result = enqueue_notify(client, client_id, {\n action: \"DATA\",\n key: action.key,\n data: action.value\n });\n });\n return enqueue_notify(client, client.client_id, {\n request_id: action.request_id,\n status: \"ok\"\n });\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":863,"value":"put_operation_response = viewof run_put_operation_args.respond(\n run_put_operation_effect\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1257,"value":"### run_get_operation\n\nA GET operation retrieves the data at a data location.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1261,"value":"viewof run_get_operation_args = flowQueue({\n name: \"run_get_operation_args\"\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1263,"value":"run_get_operation_args","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1265,"value":"run_get_operation_effect = {\n const client = run_get_operation_args.client;\n const action = run_get_operation_args.action;\n const data = run_get_operation_args.data;\n return enqueue_notify(client, client.client_id, {\n request_id: action.request_id,\n status: \"ok\",\n data\n });\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1267,"value":"get_operation_response = viewof run_get_operation_args.respond(\n run_get_operation_effect\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":601,"value":"### run_listen_operation\n\nA LISTEN operation adds a client to the list of listeners at a data location. The client is also informed of the current value of the data location.\n\nTODO: Firebase also supports query filters and listening over collections of location too.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":621,"value":"viewof run_listen_operation_args = flowQueue({\n name: \"run_listen_operation_args\"\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":627,"value":"run_listen_operation_args","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":694,"value":"run_listen_effect = {\n const client = run_listen_operation_args.client;\n const action = run_listen_operation_args.action;\n const data = run_listen_operation_args.data;\n enqueue_notify(client, client.client_id, {\n action: \"DATA\",\n key: action.key,\n data: data\n });\n add_data_listener(client, action.key);\n return enqueue_notify(client, client.client_id, {\n request_id: action.request_id,\n status: \"ok\"\n });\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":647,"value":"listen_operation_response = viewof run_listen_operation_args.respond(\n run_listen_effect\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":734,"value":"### run_unlisten\n\nUnlisten removes a listener at a data location.","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1325,"value":"viewof run_unlisten_operation_args = flowQueue({\n name: \"run_unlisten_operation_args\"\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1327,"value":"run_unlisten_operation_args","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1332,"value":"run_unlisten_effect = {\n const client = run_unlisten_operation_args.client;\n const action = run_unlisten_operation_args.action;\n remove_data_listener(client, action.key);\n return enqueue_notify(client, client.client_id, {\n request_id: action.request_id,\n status: \"ok\"\n });\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1334,"value":"unlisten_operation_response = viewof run_unlisten_operation_args.respond(\n run_unlisten_effect\n)","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":4305,"value":"$4c","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":4308,"value":"### Follow along\non [Twitter](https://twitter.com/tomlarkworthy) or \n","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":596,"value":"---","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":588,"value":"## Notebook Dependencies","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":3992,"value":"import { toc } from \"@nebrius/indented-toc\"","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":4228,"value":"import { gfx } from \"@tomlarkworthy/hackable-realtime-database-title-graphic\"","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":544,"value":"import { flowQueue } from \"@tomlarkworthy/flow-queue\"","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":4244,"value":"import { createSuite, expect } from \"@tomlarkworthy/testing\"","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1681,"value":"### Notebook Backups and Analytics","pinned":false,"hidden":false,"mode":"md","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"","database":"$undefined"},{"id":1679,"value":"import { footer } from \"@endpointservices/footer-with-backups\"","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"},{"id":1686,"value":"footer","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","tool":"$undefined","providerMetadata":"$undefined","attachments":"$undefined","output":"$undefined","database":"$undefined"}],"files":"$W4d","stdlib":"1"}} 44:"$43:body"