This is very well written. Everything I could think of on a first read has already been addressed.
One minor suggestion is to change this wording to be clearer (it has been used in 2 places):
Distributed Transaction - VTGate restart after transaction prepared on partial RMs -> Distributed Transaction - VTGate restart after transaction prepared on a subset of RMs

Just some thoughts and questions -

With respect to Distributed Transaction Identifier (DTID) Generation. In proposal 3 why do we need a nano ID? If we are going to get the DTID from the MM, then can't it just use auto-incrementing numbers after the keyspace shard? Since the transaction will be written to the MM's mysql table for the creation to be successful. We can initialize the initial value with 1 more than the max available in the said table. This will prevent collisions even across restarts, reparents, or any other issues.

For Transaction Resolution Design, In proposal 2, we can get VTOrc to just run the Resolution code too. It is only going to entail calling RPCs against vttablet. If we create a common function in a package, we can get VTOrc to do it too. For example, we don't get VTOrc to tell vtctld to run PRS/ERS, it just runs it. The same concept here can be used.

For Prepared Transactions Connection Stability. I'm just summarizing what we already discussed before. If we decide to go ahead with Proposal 1, then we need to make sure that the Unix socket connections can't be dropped by MySQL. If for whatever reason MySQL drops a connection that is running a distributed transaction, any other write that was blocked on the same rows might go through, and then the distributed transaction's state would be unrecoverable.
There is a proper fix for read and write exclusion but it entails doing locking at a level higher than MySQL connection using a shadow table as implemented by dropbox. That has a significant performance penalty however.
Edit: ☝️ is essentially the same as proposed in the next section's first proposal too. We would just overload the shadow table for also read/write exclusion.

Could you explain case 4? Why do we rollback the transactions on RM1 and RM2? There was no transaction preparation done before this, right?

Could you explain case 4? Why do we rollback the transactions on RM1 and RM2? There was no transaction preparation done before this, right?

We still have open transactions so we should roll them back otherwise they will remain holding locks till transaction timeout is not achieved.

Me and @harshit-gangal were looking at what we would need to do to integrate atomic transactions into SwitchTraffic and OnlineDDL workflows.

We found that both SwitchTraffic and OnlineDDL use the Query Rules stored in the QueryEngine field queryRuleSources.
SwitchTraffic updates this by changing the DeniedTables in ShardInfo and then asking the vttablets to refresh their state. OnlineDDL engine has the advantage that is also lives on the vttablet and doesn't require a RPC, it can make the changes by calling a local function.

Changes to these query rules govern how vttablet would run incoming queries. There is one gotcha here though, currently query rules don't impact the queries that have already run and are part of a transaction. So, when these transactions are committed, there is a possibility of going against the query rules even after they've been updated.
SwitchTraffic tries to address this race by trying to run LOCK TABLES twice giving the transactions some more time to finish running.

This is a problem for atomic transactions because once we have prepared a transaction and the transaction manager decides to commit it, then we cannot refuse. So, for example, if a transaction is in a prepared state, and then online ddl changes the rules and proceeds with the cutover, the transaction might not be able to commit because of a structural change to the underlying table.

Because both the operations are relying on query rules to communicate with the query server on what kind of queries to allow and deny, we think it's a good idea to use them for the transactions going into the prepared state too. Here is a timing diagram with the order of operations that we think will ensure that no atomic transaction interferes with the workflows -

sequenceDiagram
    participant SwitchTraffic
    participant vttablet
    participant onlineDDL

    Note over SwitchTraffic: Update Denied<br/>Tables in Vschema
    Note over onlineDDL: Register new Query Rules
    SwitchTraffic-)vttablet: RefreshState

    Note over vttablet: Refresh ShardInfo<br/>From Vschema
    Note over vttablet: Update Query Rules
    Note over vttablet: All new transactions while<br/>being Prepared will consult<br/>new query rules
    vttablet-)SwitchTraffic: Success

    SwitchTraffic-)vttablet: WaitForPrepared
    onlineDDL-)vttablet: WaitForPrepared
    Note over vttablet: Wait for all current set<br/>of prepared transactions<br/>to succeed.
    Note over vttablet: New transactions are<br/>guaranteed to be using<br/>the new query rules.
    vttablet-)SwitchTraffic: Success
    vttablet-)onlineDDL: Success


    Note over SwitchTraffic: Proceed
    Note over onlineDDL: Proceed

Atomic Transactions with PRS and MySQL Restarts

Current state and the problems

I have been looking at the code that tries to redo the prepared transactions in case of a MySQL restart or PRS, etc, and have found a few shortcomings.
To start with, the order of operations that we follow when we promote a new primary has a race condition.

1. We start by calling ChangeTabletType.
2. This call sets MySQL to read-write.
3. When we transition to the primary state, we open the query engine first as part of sm.connect in servePrimary.
4. After this step we open the transaction engine, and it starts the prepareFromRedo function in a go routine.

Because we turn off MySQL read-only and open the query engine first, we can potentially have a write that goes through before the transaction engine has had a chance to try and redo the prepared transactions. This can cause the prepare to fail since the write might be incompatible with the ones that the prepared transaction is running.

This problem is also present when MySQL restarts.

1. When MySQL restarts, all current connections return a failure and it triggers the codepath in checkMySQL.
2. We enter the non-connected state that closes all engines including transaction engine.
3. This code too then sets the wanted state to serving and when we transition, we end up calling servePrimary which has the same race described above ☝️.
4. It is also worth noting that when MySQL restarts, it starts with super-read-only turned on, and its VTOrc that fixes this by calling UndoDemotePrimary.

Solutions that were tried but didn't work

Me and @harshit-gangal have talked about this issue and I've thought off and tested a few different solutions. These are some of the solutions I tried but didn't work (If this is not interesting to read, then you can skip this segment and go to the proposed solution) -

1. One of the things that we thought we can do is that before we turn off read only in MySQL globally, we try to redo the prepared transactions and while doing so we turn off read only in that session. This would have resolved the race, because this would have ensured that no writes could have gone through until the prepared transactions were restored. This however had to be scrapped because read_only is not a session setting.

mysql [localhost:8033] {msandbox} (performance_schema) > set session read_only = 'false';
ERROR 1229 (HY000): Variable 'read_only' is a GLOBAL variable and should be set with SET GLOBAL

2. MySQL allows running FLUSH TABLES WITH READ LOCK even in the super_read_only state and this takes a lock on all the tables. So like the previous solution before we turn super_read_only off, we can take run the mentioned command to lock all the tables, and then turn off read_only and proceed with the prepared transaction and then UNLOCK TABLES. This however doesn't work because FLUSH TABLES doesn't let you acquire a write lock, so even if we lock the tables, we can't write to them...

mysql [localhost:8033] {msandbox} (test) > INSERT INTO t1 values (234);
ERROR 1223 (HY000): Can't execute the query because you have a conflicting read lock

3. At this point, I tried using a LOCK TABLES ... WRITE command instead of the flush tables command above, and I had high hopes that this would indeed work for us because lock tables acquires a write lock. But unfortunately this doesn't work either because as per MySQL docs -

If a session begins a transaction (for example, with START TRANSACTION), an implicit UNLOCK TABLES is performed, which causes existing locks to be released.

So even if we acquire the lock, we can't start the transaction to restore the state. Also we can't acquire the table lock after we start the transaction because again as per docs -

LOCK TABLES is not transaction-safe and implicitly commits any active transaction before attempting to lock the tables.

Proposed solutions that should work

1. We will store the last time the transaction engine restored the prepared state. When a new connection is being created to MySQL, we can check the MySQL uptime (which will tell us when MySQL restarted) against the stored time of last restore to infer whether it is safe to allow this connection to run writes or should we delay it.
   This solution works because we are indirectly preventing any writes to MySQL until the transaction engine has been able to redo the prepared transactions by preventing any new connections from being created. There will be some performance penalty because we will have to do this check everytime we create a new connection, but it shouldn't be too bad because of connection pooling.

2. Before we turn off read_only in MySQL, we do the following -
   a. Turn off super_read_only but leave read_only on. This will continue to prevent any writes from the users from going through.
   b. Use the dbaPool that has admin privileges to restore the prepared transactions.
   c. Turn read_only off and start all the other engines.
   This solution relies on no writes going through from the dbaPool. It is only meant to be used for replication-related queries and to kill queries. This is an invariant that we'll need to verify and in case there are DMLs using this pool, we can move them to the appPool instead.

Of the 2 proposed solutions, we like the second one more. It is less hacky and it feels like we are less likely to run into any unexpected issues with this fix.

We still need to discuss what should we do in case the redoing of transactions fails. It shouldn't happen with proposed fixes ☝️, but if it does, then what should we do? Should we panic the vttablet to prevent any writes? Or should we just setup metrics for this so that the user is alerted but allow the vttablets and other writes to function?

EDIT: We decided to go ahead with solution 2. The expectation is that the default mode that Vitess uses on replicas (super_read_only) is not being overridden. Atomic transactions are reliant on the guarantees provided by super_read_only and assume that all writes are being done using an application user that does not have admin privileges. Also, we should check that the durability policy is semi-sync when atomic transactions are being used.

In case of a failure, we want to allow vttablets to continue to take writes. We will add metrics around this so that users can get notified of these failures. We want to build vtctld commands to be able to see and resolve atomic transactions. These should then be used to implement the same functionality in vtadmin so that users can resolve transactions using the UI as well.

Task List

• Implement solution 2 - Atomic Transactions correctness with PRS, ERS and MySQL & Vttablet Restarts #16553
• Verify Atomic transactions can handle MySQL restarts, PRS and ERS using the fuzzer - Atomic Transactions correctness with PRS, ERS and MySQL & Vttablet Restarts #16553
• Verify semi-sync durability policy is being used when running atomic transactions - Reject TwoPC calls if semi-sync is not enabled #16608
• build vtctld command to see resolved transactions and take necessary steps Add conclude transaction command to vtctld service #16693 Add Unresolved Transactions command in vtctld service #16679
• build vtadmin integration for ☝️

Atomic Transactions with OnlineDDL

What Online DDL does right now

I have looked at the Online DDL code and it is only the cutover step that needs to coordinate with atomic transactions. As it stands now, Online DDL does the following steps during a cutover (More details available at https://vitess.io/blog/2022-04-06-online-ddl-vitess-cut-over/) -

1. It waits until vreplication has finished the copy phase and the lag between the table and the shadow table is not that high.
2. The first thing we do is that we start buffering of queries for the given table in question. This is done by using query rules that exist in query engine. These are described in more detail in RFC: Atomic Distributed Transactions #16245 (comment). This affects all the new queries that are coming in, but if there are open transactions which have already executed some DMLs against the table, then they can still commit. Also it can happen that the check for query rules for a query happened right before they were updated, so it might still run.
3. Online DDL then wait for 100 milliseconds to allow for queries that would have passed the query rules check before they were updated to complete. This step is not crucial for correctness, it just prevents some user errors. It can still happen though that a user query doesn't run in this time frame and runs when the RENAME has already started. In which case it will momentarily block and on unblocking it would find that the table no longer exists.
4. If the force cutover has been specified, then online DDL goes ahead and actively kills all the queries and transactions that are holding any locks against the table. This can have false positives too (code comment says so). To find the connections that are holding the locks, we are using information_schema.processlist, performance_schema.data_locks and information_schema.innodb_trx tables.
5. Online DDL now acquires write locks on both the original and the sentry table.
6. It then proceeds to start a go routine to perform the renames of the tables which gets blocked because the other connection is holding the locks.
7. After that we get the MySQL position after the lock and wait for vreplication to finish processing all the updates till that point and then we stop vreplication.
8. Then we unlock the tables and wait for the rename to complete. (original table to a sentry table, shadow table to original, and then sentry to shadow table, effectively swapping original and shadow table)
9. At this point online ddl cutover is complete, we asynchronously reload the schema and reenable the writes that were previously buffered.

From the safety of online-ddl standpoint is concerned, the buffering and then acquiring locks on writes is sufficient to guarantee that we don't lose any writes that happen on the original table when we swap.
However, the force cutover step that kills the transactions holding locks is going to cause issues for atomic transactions. For an atomic transaction that has entered the prepared state, we cannot under any circumstances kill it, because if the transaction manager decides to commit it, then we have to abide by that decision.

Proposed changes

Basically after we start the buffering and before we kill the transactions, we'll just need to wait for the prepared pool to have finished all the transactions present at that point in time. And we need to do something special for transactions trying to enter the prepared state. We have two alternates described below. Then we should be safe to kill all the remaining transactions (atomic or not).

I'm hoping this will work quite nicely in terms of timing, because prepared transactions should really be committed/rollbacked pretty quickly. Transactions are in the prepared state only after the user has already run the commit command on vtgate, and we have completed the first part of 2pc. So if things are not broken (mostly they won't be), then it should be a matter of less than a second for it to receive the result of the decision (commit/rollback). So, unless something catastrophic happens that leave a transaction in the prepared state and we need the resolution to happen by a transaction watcher, the transactions in prepared state would be very short lived. Waiting is different for a transaction that the user has currently "open" (in the sense they haven't committed it), so the user can run more DMLs. We don't need to wait for them. They can be killed. Atomic transaction guarantee is based on the premise that prepared transactions can't fail, and we only enter the prepared state after the transaction has reached a conclusion from the user perspective (they have issued commit).

Another thing that we need to consider apart from waiting for all prepared transactions to go through, is whether we allow other transactions to enter the prepared state -

1. The safest and simplest to implement solution would be to not allow any transaction to enter the prepared state while the online-ddl cutover is ongoing. Much like how we buffer the queries for the table, we can buffer all the requests for entering the prepared state. When we try to enter the prepared state, we just need to then check if the connection was killed or not. If it was then we fail the prepare. The disadvantage of this approach is that we will stall atomic transactions that are not affecting this table too. Even though the cutover is a pretty fast process, the delay will be visible in the transaction commit latencies.
2. The other option that we have is that we check which tables a transaction is accessing when it enters the prepared state, and we check the query rules again. If we have a query rule that asks to buffer or reject queries for a table that is being used, then we fail the prepare request. This approach allows us to continue preparing transactions that don't affect the table that is part of the online DDL. We will have to ensure that the killing logic for transactions doesn't accidentally end up killing any of the new prepared transactions.

Other Online DDL variants (ghost and pt-osc)

The above discussion was all limited to online DDL being run using Vreplication. This is the only mode in which we have any control over the cutover phase. Both ghost and pt-osc are running as separate binaries, so it is going to be very hard to collaborate with either of them. We need to ensure that neither of them is killing transactions during their cutover phase. If they are, then it is going to be impossible to guarantee atomic transactions with them. Even if they aren't killing transactions, it might be very hard for them to find a window for running the cutover if the system is under enough write load because they won't be able to acquire the locks on the table.

UPDATE: ghost has been deprecated and pt-osc has always been experimental. #15692 We can just state that their limitations with atomic transactions and move on.

UPDATE: We decided to go ahead with the second solution. We will need to make sure we handle all races correctly because it could happen that the check for query rules in prepare state happens before we update the rules, but it takes some time for the transaction to show up in the prepared pool, so we could potentially have a prepared transaction not in the list of transactions that we wait for, but still using the table in question. To handle this, we need to be defensive in what we kill in the online DDL segment. We need to consult the prepared transactions pool to make sure we don't kill something that is prepared.

Task List

• Implement solution 2 - Atomic Transactions handling with Online DDL #16585
• Online DDL Kill logic should be defensive wrt to what it kills. Check prepared pool to make sure we don't kill prepared transactions.
• Check and verify races are handled. - Atomic Transactions handling with Online DDL #16585
• Verify the additional load on vttablets because of atomic transactions with online DDL doesn't cause it to OOM (running atomic transactions will slow down online ddl so it could cause more queries to be buffered, and thus an OOM)

Atomic Transactions with MoveTables and Reshard

Order of operations for MoveTables and Resharding

The only operation that we are concerned about wrt to atomic transactions is the switch write. The following is the order of operations for Switch Write -

1. Read the workflow state that entails reading the shard and tablet records to find the shards that have already switched writes.
2. Run validations that the list of shards between the source and destination matches.
3. Lock the keyspace for both the source and target.
4. We check the metadata for sequence tables and then whether we already have the journal entry. If we do, then a previous switch write would have already run a few of the following steps, so we skip them.
5. We read information about existing streams running on the shards, and stop them.
6. Next we stop writes on the source and this is the step that concerns atomic transactions. It is implemented slightly differently for both Resharding and MoveTables -
   a. For MoveTables, we want to disallow writes to only the tables involved. We use DeniedTables in ShardInfo to accomplish this. Both vtgate and vttablet use the DeniedTables. In vtgate, it is used in the keyspaceState to generate keyspace events to know when to stop buffering. On vttablets, it is used to set the queryRules as described previously in RFC: Atomic Distributed Transactions #16245 (comment). These query rules are used to reject any further DML against the tables and the returned error starts buffering on vtgate. After we update the topo server with the new DeniedTables, we make all the vttablets refresh their topo to ensure that they've registered the change. We also call RebuildSrvVschema which causes vtgates watcher on the srv vchema to cause it to recheck the shard states. The updation of denied tables on vtgate is however not synchronous like vttablets.
   b. For Resharding, we want to disallow all writes for the given shard range, so we set DisableQueryService in the srvVschema. This field too is something that vttablets monitor. And like before, after we set the disable query service field, we get vttablets to refresh the information from topo. The field is used to set tabletControls in the tablet state which is eventually used to transition the tablet into a non-serving state, effectively blocking any queries. In StartRequest , which encapsulates all DMLs and transactional calls etc, checks for this serving state before allowing any query.
7. Once we have prevented new writes, we get the MySQL positions of the primary tablets.
8. Next, for MoveTables, we try Locking the tables to ensure no existing write is pending. We do this twice to allow for races between the denied tables being set and some writes still pending. (This part I was a little confused about. Shouldn't we be getting the primary position after we have executed the lock on the tables?)
9. We wait for vreplication to catch up to the position we stored previously.
10. Next we migrate all the streams (information for which we stored previously). Similarly we handle sequences and then start reverse vreplication.
11. We create the journals, and allow writes on the target. This is the exact opposite of what we do for stopping writes. i.e. we remove the denied tables (for MoveTables) and DisableQueryService (for Resharding)
12. Finally, we update the routing rules to allow vtgates to direct traffic to the target.

Proposed Changes

We propose a similar fix to the one proposed for onlineDDL in #16245 (comment).

On the vttablets we can always tell if a reshard or moveTables is in progress because of the fields that they set (DisableQueryService, and DeniedTables respectively). So, after we have updated the said field, we'll just need to wait for the prepared pool to have finished all the transactions present at that point in time. And like before, we will have to prevent new prepared transactions from being prepared by consulting these fields -

1. For MoveTables, the change we make for online DDL will work too, because it updates the query rules.
2. For Reshard, we will have to check the tabletControls field to check if resharding is in progress.

The same race conditions that we need to handle for online DDL stand here too. One thing that is different though, is that we'll have to introduce a new RPC for getting vtctld to wait for the prepared transactions to finish on the vttablet side. In the case of OnlineDDL the executor was also running on the vttablet so no RPC was required. But in this case we will need to implement something like WaitForPreparedTransactionsToRoll.

Another important consideration is that the way that the code is written currently, for CommitPrepared (function we call to commit a prepared transaction), also goes through the StartRequest flow. If we don't fix this, we would end up in a situation, wherein Resharding is waiting for prepared transactions to roll (as stated above), but those transactions won't be able to be committed because startRequest will continue to block the CommitPrepared call. We need to allow running CommitPrepared even in the case where the tablet is not serving and then check for correctness in that call later.

Update: We will go ahead with the proposed fix. We discussed a few other considerations that we might run into with Resharding and CommitPrepared not being allowed in the non-serving state. Me and Harshit will talk about this more after looking at the code.
Another important thing to note is that atomic transactions won't work for shard-by-shard migration or tenant-based migration. If the unmanaged tablet being used for them is connected to the MySQL instance with a TCP connection, then it will reject any call to prepare the transaction.

Update: We went with an alternate way of making atomic transactions work with Resharding. Instead of doing the changes proposed ☝️, we instead went ahead and changed RefreshState RPC that Reshard uses. In that call, when the tablet sees that it should disable query service, it first turns off the prepared pool from accepting any new prepared transactions and waits for the existing prepared transactions to conclude. This ensures that Reshard only proceeds when all the prepared transactions have been resolved and there are no prepared transactions that are pending.

Task List

• Changes to make MoveTables work with 2pc - Move tables with atomic transactions #16676
• Changes to make Reshard work with 2pc - Making Reshard work smoothly with Atomic Transactions #16844

Here are the state transitions for an atomic transaction on a single resource manager as me and @harshit-gangal discussed -

---
title: State Changes for each participant in an atomic transaction
---
stateDiagram-v2
    classDef action font-style:italic,font-weight:bold,fill:white
    state "Redo Prepared Logs" as rpl
    state "Delete Redo Logs" as drl
    state "Check Type of Failure" as ctf

    state "Prepared" as p
    state "Failure" as f
    state commitRPC <<choice>>
    state failureType <<choice>>
    state redoSuccess <<choice>>

    [*] --> p: Commit redo logs<br>Prepare Call Succeeds
    drl --> [*]
    p --> drl:::action: Receive Rollback<br>Rollback transaction
    p --> commitRPC: Receive CommitRPC
    commitRPC --> drl: If commit succeeds
    ctf --> failureType
    commitRPC --> ctf:::action: If commit fails
    failureType --> f: Unretriable failure<br>Wait for user intervention
    f --> drl: User calls ConcludeTransaction
    failureType --> rpl:::action: Failure because of MySQL restart
    rpl --> redoSuccess: Check if redo prepare fails
    redoSuccess --> p: Redo prepare succeeds
    redoSuccess --> ctf: Redo prepare fails

• As an extra TODO item, we need to also wait for atomic transactions to be rolled when a user externally changes the query rules.