Merge #307

307: sqld programs r=MarinPostma a=MarinPostma

This PR implements sqld programs on the model of hrana programs.

Programs take over batches in the execution of queries within sqld, since they are very flexible and let the `Database` API consumer specify how statements should be executed.

The API of the sqld programs closely matches that of hrana batches, so the implementation on hrana side is changed to simply transform a hrana batch into a sqld program.

On the HTTP side, batches of queries are transformed into a program that causes a rollback if any statement in the batch fails.

On the replica, the programs are serialized and sent over by gRPC to the primary when we detect that they could cause a write. 

This PR also solves a known bug in HTTP batches that used to unconditionally execute statements in a batch even if a statement in the batch failed, and never rolledback, since the new implementation of HTTP batches ensures that a rollback is always performed if any statement in the batch fails to execute.

Future work: expose a more direct access to the `Database` to the consumers

Co-authored-by: ad hoc <[email protected]>

Author: bors[bot]

sqld/proto/proxy.proto

@@ -75,10 +75,10 @@ message Column {
 }
 
 message DisconnectMessage {
-    bytes clientId = 1;
+    string clientId = 1;
 }
 
-message Ack {}
+message Ack { }
 
 message ExecuteResults {
     repeated QueryResult results = 1;
@@ -91,7 +91,51 @@ message ExecuteResults {
     State state = 2;
 }
 
+message Program {
+    repeated Step steps = 1;
+}
+
+message Step {
+    optional Cond cond = 1;
+    Query query = 2;
+}
+
+message Cond {
+    oneof cond {
+        OkCond ok = 1;
+        ErrCond err = 2;
+        NotCond not = 3;
+        AndCond and = 4;
+        OrCond or = 5;
+    }
+}
+
+message OkCond {
+    int64 step = 1;
+}
+
+message ErrCond {
+    int64 step = 1;
+}
+
+message NotCond {
+    Cond cond = 1;
+}
+
+message AndCond {
+    repeated Cond conds = 1;
+}
+
+message OrCond {
+    repeated Cond conds = 1;
+}
+
+message ProgramReq {
+    string client_id = 1;
+    Program pgm = 2;
+}
+
 service Proxy {
-  rpc Execute(Queries) returns (ExecuteResults) {}
+  rpc Execute(ProgramReq) returns (ExecuteResults) {}
   rpc Disconnect(DisconnectMessage) returns (Ack) {}
 }

sqld/src/batch/mod.rs

@@ -23,7 +23,7 @@ impl DumpLoader {
 
         let (ok_snd, ok_rcv) = oneshot::channel::<anyhow::Result<()>>();
         tokio::task::spawn_blocking(move || {
-            let db = match open_db(path, hooks, false) {
+            let db = match open_db(&path, hooks, false) {
                 Ok(db) => {
                     let _ = ok_snd.send(Ok(()));
                     db