Keeper cross-shard transactions

Just a design proposal.

### Boring philosophising

Suppose we want lots of tables that can talk to each other (but not very frequently). E.g. a bigger clickhouse instances, or tables shared across multiple instances. We'd need to change something about how keeper is used. Here's one thing we could (probably) do.

Have multiple keeper clusters. Allow different tables to live on different keeper clusters. Allow different ch instances to share tables by sharing the corresponding keeper clusters. The clusters would be mostly normal independent keeper clusters. But occasionally we'd need to do e.g. `MOVE PARTITION` from one table to another. `MOVE PARTITION` would need to somehow make sure it never loses or duplicates the partition if it crashes in the middle of the operation. I can think of two ways of going about it:
1. Come up with some ad-hoc way of doing this for each such type of query. E.g. maybe `ATTACH PARTITION` would create a znode describing this pending operation in each table, then after crash it would check for such znode and resume or clean up somehow.
2. Or modify keeper to support cross-shard transactions. Multiple keeper clusters would talk to each other and coordinate a proper consistent all-or-nothing conditional multi-write.

Option 2 is kind of just option 1 but built into keeper and implemented only once. I don't know which is better. This writeup explores option 2.

(EDIT: Maybe this is not enough: if we want to be able to share data between tables through hardlinks, s3_with_keeper disk would have to somehow support having hardlinks to the same S3 object in different keeper clusters. I guess this can also be done with cross-shard transactions, but we'd need to be careful about performance.)

### Cross-shard transaction design

Suppose:
* there are a few normal independent keeper clusters,
* they all know about each other; maybe they share a config file that describes all the clusters,
* each cluster is called a "shard" and has a shard id.

And we want this new API:
```
struct MultiRequestForShard
{
uint64_t shard_id;
Coordination::Requests requests;
};

/// Run requests on multiple shards. Either all or none of the requests succeed.
Responses crossShardTransaction(vector requests_for_shards);
```

And we don't need it to be very fast. Just a few cross-shard transactions per second (in addition to a ~normal throughput of normal single-shard operations). (I don't know how to do high-throughput cross-shard transactions with zookeeper-like data model.)

Here's one way to do it. It's my favorite flavor of two-phase commit.

Tl;dr: a cross-shard transaction takes two raft roundtrips of latency, during which time all znodes involved in the transaction are "locked", i.e. other requests can't read or write these znodes.

Short version:
* In the keeper state machine's state, next to the map of znodes etc, add wall-clock timestamp of last committed entry. Enforce that is monotonically increases with each entry (e.g. `last_entry_time = max(last_entry_time + 1, now)`) So keeper now has an official answer to the question "what time is it?".
* To the same state, and add a set of pending cross-shard transactions. Each such tx has:
* the list of participating shards and a list of requests for each of them
* one of these shards is designated as "coordinator"
* status, one of: `pending`, `committed`, `failed`
* deadline, wall-clock time; by definition, a tx is considered successful if all participating shards add it to their pending set (in non-failed status) before this deadline (according to the official time as above, so this doesn't rely on clock synchronization for correctness, it is not spanner)
* a uuid
* And also add a set of "locked" znode paths: znodes that are touched by any tx with `pending` status. All reads and writes for those znodes are rejected. Because the result of such reads would depend on whether the tx succeeds, and we don't know whether it will succeed yet. (And other cross-shard transactions touching those znodes are rejected, but transactions for other znodes are fine.) (And add a separate set of the parents of touched znodes, to similarly block `getChildren` requests that depend on tx outcome.)
* (Note: below, "messages" here means just new types of keeper requests, just like `create` or `getChildren` or whatever. They just modify different parts of the keeper state machine's state. "Send a message" means run a request using keeper client, "get a response" means completion of that request.)
* When starting the transaction, the coordinator sets its deadline to some time a few seconds in the future. Then sends a "prepare" message to all participating shards, including itself, in parallel. Recipient adds the tx to its pending set with either `pending` or `rejected` status (rejected if the requests fail, e.g. if there's a znode version check request). This operation is conditional on the add-tx-to-pending-set raft entry having a timestamp below the tx's deadline; if we're too late, the write is rejected. (So, if we ever see that keeper's state has `last_entry_time > deadline`, but doesn't have the tx in its pending tx set, we know this tx will never be added to the pending set.)
* After coordinator gets all responses, it knows the outcome of the tx: iff any responses are rejections, the tx should be considered failed. (Note that unavailability (e.g. timeout or connection loss) doesn't count as a response; "prepare" has to be retried until a success or rejection is received.)
* Coordinator then sends "commit" or "fail" messages to all participating shards, including itself, in parallel. Recipient changes the tx's status to `committed`/`failed` and, if `committed`, applies the tx's requests to its actual znodes state.
* Thus we have applied a cross-shard transaction in 2 roundtrips of latency (first "prepare" on all shards in parallel, then "commit" on all shards in parallel).
* Now we just need to eventually clean up the completed or failed transactions from the pending tx set. It can be done with another two roundtrips (which can be piggy-backed to later operations since there's no hurry). Like this. After the coordinator's own state was updated to have the tx in `committed` or `failed` status (as part of the broadcast from previous step), coordinator can send "commit and delete" or "delete" (if `failed`) messages to everyone except itself. After receiving all acks for that (i.e. everyone else has removed the tx from the pending set), it can send "delete" to itself as well. (Note: it's "commit and delete" instead of "delete" because if at this point we're not sure that the previous "commit" was delivered; this cleanup can start as soon as the "commit" is stored on the coordinator, without waiting for everyone else. This also unifies the logic between normal coordination and recovery after failure: coordinator can just always send out "commit and delete" if it sees "committed" locally.)
* Recovery procedures are left as an exercise to the reader. When a keeper leader comes up, it should check its pending tx set and do the coordinator stuff for the txs for which it's the coordinator, and not do anything for txs where it's not the coordinator (relying on the coordinator to drive its txs to completion, with retries and all).

Long version: ... I can write it on demand if there's any interest in this idea. (Or maybe there's no need to write long versions anymore because anyone can ask chatgpt to provide context? Make sure to use the smartest model around, e.g. Claude 4.6 failed to understand some version of this on one attempt.)

### Improvements

#### Delaying instead of failing requests for locked znodes

As described above, leader has to briefly reject (at pre-append stage) requests that read or write "locked" znodes involved in active cross-shard transactions. To avoid inconveniencing users with those rejections, we could add some kind of retries for such rejected requests. It can be done cleanly (without reordering requests from the same session), but I don't know what exactly the implementation should look like, seems tricky. (I guess the new KeeperDispatcher from https://github.com/ClickHouse/ClickHouse/pull/101757 can roll back the whole stream (so requests from other sessions would also be delayed, unnecessarily) on getting such rejection. Or KeeperAppendStream could do some more complicated per-session tracking of errors, or we could do something else, idk, I don't see a clear winner yet.)

#### Run multiple shards on the same machine

Keeper super underutilizes cpu because its data model is inherently single-threaded, so we're using just a few cpu cores. Usually we can't run multiple keepers on the same machine because they use lots of memory (for keeping all those znodes). But with sharding, each shard can be smaller, and we can run multiple shards on the same machine and potentially use all cpu cores.

#### Maintain full state only on leader

Now that we can use more cores, the ratio of usable-cores to usable-memory has increased. This means we'll be more memory-bound. (Uh, this sounds like it contradicts the previous section? I'm still not sure how to think about it.) And we don't need as much throughput from each shard. So we can cut memory usage by 3x by keeping the znode map in memory only on leader. All client requests would go to leader. Followers would just store the changelog without interpreting its contents. Snapshots would either go to s3 or be periodically sent to followers to store on disk. This decreases memory usage by 3x, but decreases single-shard throughput (but not total throughput) by something between 1x-3x (because followers can't execute reads). This is a good tradeoff in this setup.

### Conclusion

In conclusion, hotdog is a sandwich.