Fix invalid state in transaction (#2009)

Fix #2004.

The gRPC code gets the `Status` information from the server to know if a
transaction was successful. Now, every time we try to run a statement,
the `is_autocommit` is run on the remote connection to know if we are in
a `State::Txn` or not.

After that, the code is basically a copy of
`crate::replication::connection::should_execute_local` with the addition
that when matching if the code should execute locally, the `state` is
set to the `predicted_end_state`. This step was skipped by the old
implementation because after a remote statement statement executed, the
end state came from the server as mentioned before.

`state` is mostly concerned to know if the server is in a transaction
and that we should keep sending remote queries until a `COMMIT` or
`ROLLBACK`. The match statement _could_ be refactored to reflect this
more clearly.

Author: penberg

libsql/src/sync/connection.rs

@@ -3,6 +3,7 @@ use crate::{
     hrana::{connection::HttpConnection, hyper::HttpSender},
     local::{self, impls::LibsqlStmt},
     params::Params,
+    parser,
     replication::connection::State,
     sync::SyncContext,
     BatchRows, Error, Result, Statement, Transaction, TransactionBehavior,
@@ -33,7 +34,66 @@ impl SyncedConnection {
 
         let mut state = self.state.lock().await;
 
-        crate::replication::connection::should_execute_local(&mut state, stmts.as_slice())
+        if !self.remote.is_autocommit() {
+            *state = State::Txn;
+        }
+
+        {
+            let predicted_end_state = {
+                let mut state = state.clone();
+
+                stmts.iter().for_each(|parser::Statement { kind, .. }| {
+                    state = state.step(*kind);
+                });
+
+                state
+            };
+
+            let should_execute_local = match (*state, predicted_end_state) {
+                (State::Init, State::Init) => stmts.iter().all(parser::Statement::is_read_only),
+
+                (State::Init, State::TxnReadOnly) | (State::TxnReadOnly, State::TxnReadOnly) => {
+                    let is_read_only = stmts.iter().all(parser::Statement::is_read_only);
+
+                    if !is_read_only {
+                        return Err(Error::Misuse(
+                            "Invalid write in a readonly transaction".into(),
+                        ));
+                    }
+
+                    *state = State::TxnReadOnly;
+                    true
+                }
+
+                (State::TxnReadOnly, State::Init) => {
+                    let is_read_only = stmts.iter().all(parser::Statement::is_read_only);
+
+                    if !is_read_only {
+                        return Err(Error::Misuse(
+                            "Invalid write in a readonly transaction".into(),
+                        ));
+                    }
+
+                    *state = State::Init;
+                    true
+                }
+
+                (init, State::Invalid) => {
+                    let err = Err(Error::InvalidParserState(format!("{:?}", init)));
+
+                    // Reset state always back to init so the user can start over
+                    *state = State::Init;
+
+                    return err;
+                }
+                _ => {
+                    *state = predicted_end_state;
+                    false
+                },
+            };
+
+            Ok(should_execute_local)
+        }
     }
 }