feat(pd): commit schema keys and a per-graph record as one raft entry

hugegraph-pd/docs/api-reference.md

@@ -474,6 +474,43 @@ for (Kv.KvPair kv : response.getKvsList()) {
 }
 ```
 
+#### Txn
+
+Atomic multi-key write: compares, ops and at most one graph record change, applied as one
+Raft entry and one RocksDB write batch. See [KV transactions](#kv-transactions) for the
+semantics and the retry contract.
+
+**Request**:
+```protobuf
+message TxnRequest {
+    pdpb.RequestHeader header = 1;
+    repeated TxnCompare compares = 2;  // key, kind (ABSENT | EQUALS), value
+    repeated TxnOp ops = 3;            // type (PUT | DELETE | DELETE_PREFIX), key, value
+    TxnRecord record = 4;              // key, op (BUMP | CREATE | DROP), expected_incarnation
+}
+```
+
+**Response**:
+```protobuf
+message TxnResponse {
+    pdpb.ResponseHeader header = 1;
+    bool succeeded = 2;
+    int64 revision = 3;      // the record's rev after the TXN (the Raft index, or the
+                             // previous rev plus one when higher); the index without a record
+    int64 incarnation = 4;   // the record's incarnation after the TXN, or the stored one
+    Failure failure = 5;     // NONE | COMPARE_FAILED | GRAPH_DROPPED | INCARNATION_MISMATCH |
+                             // INVALID_RECORD
+    int32 failed_compare = 6;
+    string state = 7;        // LIVE | DROPPED, or "" for an absent or unreadable record
+}
+```
+
+A failed compare or a rejected record is an unsuccessful response, not an error. The zero
+value of each enum is `UNSPECIFIED`: a compare, op or record whose field was never set is
+rejected before the proposal, as is an empty key, with error type `TXN_NOT_PROPOSED` (nothing
+was applied). The TXN is proposed by the leader; a follower forwards it. Each node applies the
+entry with the keys as the proposing node encoded them, so every replica writes the same bytes.
+
 ### 3. Pulse Service (HgPdPulseGrpc)
 
 Heartbeat and health monitoring for Store nodes.
@@ -756,6 +793,94 @@ for (Map.Entry<String, byte[]> entry : results.entrySet()) {
 }
 ```
 
+### KV transactions
+
+`KvClient.txn(TxnRequest)` commits writes and an optional graph record update as one Raft
+entry, applied in one RocksDB batch on every PD node. Only the leader proposes it: a follower
+forwards it. With `raft.enable=false` there is no Raft entry, leader or forwarding: the PD
+applies the txn in one RocksDB batch under its store's write lock, and every "Raft index" in
+this section is the store's own TXN count. `txn` sends the request to one PD, and once more to
+the leader found again only when the connection was refused, so the request was never sent; it
+throws `PDException` when it gets no answer, or when PD answers with a gRPC status (the status
+is in the message and the cause: `UNIMPLEMENTED` from a PD older than this RPC means the call
+was rejected before any apply, although the exception's error type is `UNKNOWN`; `UNAVAILABLE`
+and `DEADLINE_EXCEEDED` mean it was lost). A `PDException` of error type `TXN_NOT_PROPOSED` is
+PD's answer to a request it did not propose (invalid, see below, a PD peer that lacks the RPC,
+or a leader that lost the leadership before it could propose), so nothing was applied. Every
+other failure is an unknown outcome.
+The request has three parts:
+
+- **Compares.** Each names a key and a kind: `ABSENT` holds while the key does not exist (an
+  empty value is a value), `EQUALS` while its value equals the given one. If one fails, nothing
+  is written and the failure is `COMPARE_FAILED`, with the failed compare.
+- **Ops.** In order: `PUT` a key and value, `DELETE` a key, `DELETE_PREFIX` every key under a
+  prefix. An op or compare with an empty key, or whose kind, type or record op was left at the
+  zero value `UNSPECIFIED`, is rejected before the TXN is proposed. A value that was put with a
+  TTL has a cleanup marker next to it; an op that replaces or deletes such a value deletes the
+  marker in the same batch, so the TTL cleanup, which deletes the key a marker names once its
+  time has passed, cannot delete what the TXN wrote.
+- **Record.** At most one graph record, `HUGEGRAPH/{cluster}/SCHEMA_SYNC/{graphspace}/{graph}`,
+  with an op and the incarnation the writer expects:
+  - `BUMP` sets `rev` to the Raft index of the entry, or to the previous `rev` plus one when
+    that is higher (after `raft.enable` changed on the store, the index starts low again), so
+    `rev` only rises. With expected incarnation 0 it creates an absent record as `LIVE`
+    incarnation 1. It fails with `GRAPH_DROPPED` on a dropped graph and `INCARNATION_MISMATCH`
+    on another incarnation.
+  - `CREATE` starts the next incarnation, or 1, as `LIVE`, from an absent record or a valid
+    one in either state (an unreadable record fails it, as below), and marks
+    the record `"src":"create"`, which `BUMP` and `DROP` keep and nothing removes; a record a
+    `BUMP` created has no mark. A caller that must not take over a `LIVE` record guards the
+    `CREATE` with compares on its own keys and on the record as it read it (`ABSENT`, or
+    `EQUALS` the `DROPPED` value).
+  - `DROP` marks the expected incarnation `DROPPED`. It fails with `INCARNATION_MISMATCH` when
+    the record is absent or of another incarnation, and `GRAPH_DROPPED` when it is dropped
+    already.
+  - A stored record that is not `{"rev":<n>,"inc":<n>,"state":"LIVE"|"DROPPED"}` with positive
+    integers fails every txn on it with `INVALID_RECORD`.
+
+The response carries `succeeded`, the `revision` (the record's `rev` after the TXN, or the
+Raft index of the entry for a TXN without a record), the record's `incarnation` and `state`,
+and the `failure`. Keys and values share the namespace of `put`,
+`get` and `scanPrefix`, and a TXN write fires no plain watch event. PD keeps no request ids
+and a txn applied twice is not always harmless (a guarded `CREATE` fails its record compare
+the second time), so `KvClient` sends a txn once (and once more only when the connection was
+refused before anything was sent), and a PD that lost the leadership after it took a txn
+answers with the error instead of forwarding it. When `txn` throws, the outcome is unknown,
+except for `TXN_NOT_PROPOSED` and `UNIMPLEMENTED` above: read the state before retrying. This is the one call the retry advice under
+[Best Practices](#4-retry-strategy) does not cover.
+
+A TXN is proposed only once every other node of the Raft group (voters and learners) has
+answered the leader's member probe as a release that applies the TXN entry; the leader
+records the confirmed nodes, itself included, through Raft, so a new leader after a failover
+trusts them, and a peer that has not answered yet, or answered as an older release, blocks
+every `txn` with `TXN_NOT_PROPOSED` naming it, while a confirmed peer that is unreachable
+later does not block, on this leader or the next. Unconfirmed peers are probed again on a `txn`
+attempt at most every 3 seconds, and at once when the group's configuration changes, so a node
+added to the group blocks `txn` until it answers. A peer that joins applies the log, `txn` entries included,
+so a membership change (`changePeerList`, `updatePdRaft`) probes every peer it adds first and
+is refused with `TXN_NOT_PROPOSED` when one is unconfirmed or older: an old-release PD can no
+longer be added to an upgraded group. Peers the change keeps are not probed, since the check
+before every `txn` covers them, so a kept member that is down or not upgraded yet does not stop
+a membership change. Upgrade every PD node before
+the first `txn`; a PD downgrade after it is unsupported.
+
+On a restart without a snapshot, PD applies the Raft log after the last snapshot again over
+the store that holds it. Every TXN records its Raft index in the store, and a restarted node
+skips every store write at or below that index, plain puts included: a plain put from before
+the first TXN, applied again over what a later TXN wrote, would undo the TXN. Only a replay is
+skipped: an entry that was in the node's log when it started. An entry above that index is new
+in this run, and a store that claims to hold it is ahead of its log (a store restored from a
+backup without the log): the other nodes apply the entry, and this node fails the apply instead
+of skipping it, so that it does not fall behind them in silence. jraft then stops this node's
+state machine, a leader steps down, the readiness probe reports the error, and the node logs
+the entry and the stored index at error level. The same happens when the stored index is beyond
+the end of the log at start, or when the log has another term at the stored index than the one
+stored with it (the log was rebuilt or replaced while the store was kept). If the node cannot
+read its own raft log, it replays without that check and logs once, at error level, that the log
+generation could not be verified, since fencing every node on restart would leave none to rejoin
+from. Recovery: remove the node's data directory and let it rejoin from a snapshot. A PD run without Raft (`raft.enable=false`) counts its TXNs
+under a key of its own, so switching such a store to Raft does not trip this.
+
 ### KV watch recovery
 
 `KvClient.listen()` and `listenPrefix()` restore subscriptions after retryable
@@ -1057,6 +1182,9 @@ ManagedChannel channel = ManagedChannelBuilder
 - Implement exponential backoff for retries
 - Retry on transient errors (UNAVAILABLE, DEADLINE_EXCEEDED)
 - Don't retry on permanent errors (NOT_FOUND, INVALID_ARGUMENT)
+- Except for `txn`: a lost `txn` may have committed, so it is not retried once it may have
+  been sent (it goes once more only when the connection was refused before sending), and the
+  caller reads the state before retrying (see [KV transactions](#kv-transactions))
 
 ```java
 int maxRetries = 3;

hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java

@@ -18,6 +18,7 @@
 package org.apache.hugegraph.pd.client;
 
 import java.io.Closeable;
+import java.net.ConnectException;
 import java.util.LinkedList;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -42,6 +43,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.MethodDescriptor;
+import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.AbstractBlockingStub;
 import io.grpc.stub.AbstractStub;
@@ -92,15 +94,15 @@
     private static <T extends AbstractStub> T setBlockingParams(T stub, PDConfig config,
                                                                  long timeoutMillis) {
         stub = (T) stub.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS)
                        .withMaxInboundMessageSize(PDConfig.getInboundMessageSize());
         return (T) stub.withInterceptors(
                 new Authentication(config.getUserName(), config.getAuthority()));
 
     }
 
     public static <T extends AbstractStub> T setAsyncParams(T stub, PDConfig config) {
         return (T) stub.withMaxInboundMessageSize(PDConfig.getInboundMessageSize())
                        .withInterceptors(
                                new Authentication(config.getUserName(), config.getAuthority()));
     }
 
@@ -279,6 +281,88 @@
         return null;
     }
 
+    /**
+     * Sends the request once, for a call that is not idempotent: it can't be sent again when
+     * its response is lost. An UNAVAILABLE status drops the stub if it still holds the channel
+     * of the call, so the next call connects again; other failures keep it, since calls of
+     * other threads may be running on the channel and a reconnect would cut them off. A call
+     * whose connection was refused never left this client, so it is sent once more on a new
+     * stub, as a PD that stopped a moment ago leaves it.
+     *
+     * @return null when the call failed
+     */
+    protected <ReqT, RespT> RespT blockingUnaryCallOnce(
+            MethodDescriptor<ReqT, RespT> method, ReqT req) throws PDException {
+        return blockingUnaryCallOnce(method, req, false);
+    }
+
+    private <ReqT, RespT> RespT blockingUnaryCallOnce(
+            MethodDescriptor<ReqT, RespT> method, ReqT req, boolean resent) throws PDException {
+        AbstractBlockingStub stub;
+        Channel attemptChannel;
+        synchronized (this) {
+            stub = getBlockingStub();
+            // The stub of the call wraps the channel with its interceptors: compare the stored one
+            attemptChannel = proxy.getBlockingStub().getChannel();
+        }
+        try {
+            return ClientCalls.blockingUnaryCall(stub.getChannel(), method,
+                                                 stub.getCallOptions(), req);
+        } catch (StatusRuntimeException e) {
+            boolean unsent = !resent && refused(e);
+            log.warn("{} failed, {}: {}", method.getFullMethodName(),
+                     unsent ? "never sent, sending it once more" : "not sent again",
+                     e.getStatus());
+            if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
+                synchronized (this) {
+                    AbstractBlockingStub current = proxy.getBlockingStub();
+                    if (current != null && current.getChannel() == attemptChannel) {
+                        proxy.setBlockingStub(null);
+                    }
+                }
+            }
+            if (unsent) {
+                return blockingUnaryCallOnce(method, req, true);
+            }
+            throw failed(method, e);
+        } catch (RuntimeException e) {
+            log.error(method.getFullMethodName() + " exception, not sent again", e);
+            return null;
+        }
+    }
+
+    /**
+     * The failure of a call made once, with the gRPC status, so the caller can tell a call PD
+     * rejected (UNIMPLEMENTED from an older PD, PERMISSION_DENIED) from one that was lost
+     * (UNAVAILABLE, DEADLINE_EXCEEDED) and has an unknown outcome
+     */
+    private PDException failed(MethodDescriptor<?, ?> method, StatusRuntimeException e) {
+        Status status = e.getStatus();
+        boolean lost = status.getCode() == Status.Code.UNAVAILABLE ||
+                       status.getCode() == Status.Code.DEADLINE_EXCEEDED;
+        String message = method.getFullMethodName() + " failed with " + status.getCode() +
+                         (status.getDescription() == null ? "" :
+                          ": " + status.getDescription()) +
+                         (lost ? ", pd.peers=" + config.getServerHost() : "");
+        return new PDException(lost ? Pdpb.ErrorType.PD_UNREACHABLE_VALUE :
+                               Pdpb.ErrorType.UNKNOWN_VALUE, message, e);
+    }
+
+    /**
+     * Whether the call failed to connect, so its request was never written
+     */
+    private static boolean refused(StatusRuntimeException e) {
+        if (e.getStatus().getCode() != Status.Code.UNAVAILABLE) {
+            return false;
+        }
+        for (Throwable cause = e.getCause(); cause != null; cause = cause.getCause()) {
+            if (cause instanceof ConnectException) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     // this.stubs = new ConcurrentHashMap<String,AbstractBlockingStub>(hosts.length);
     private AbstractBlockingStub getConcurrentBlockingStub(String address) {
         AbstractBlockingStub stub = stubs.get(address);

hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java

@@ -49,6 +49,8 @@
 import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
 import org.apache.hugegraph.pd.grpc.kv.TTLRequest;
 import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.TxnRequest;
+import org.apache.hugegraph.pd.grpc.kv.TxnResponse;
 import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
 import org.apache.hugegraph.pd.grpc.kv.WatchKv;
 import org.apache.hugegraph.pd.grpc.kv.WatchRequest;
@@ -191,11 +193,38 @@
         return response;
     }
 
+    /**
+     * Atomic multi-key write. A failed compare or a rejected record is not an error: check
+     * succeeded on the response. The call is sent once: a TXN applied twice is not always
+     * harmless, a guarded CREATE fails its compare the second time. Only a call whose
+     * connection was refused, so never sent, goes once more to the leader found again. When
+     * no response arrives, or the PD that took the TXN lost the leadership before answering,
+     * it throws.
+     *
+     * @throws PDException with an unknown outcome: PD may have applied the TXN, except with
+     *                     error type TXN_NOT_PROPOSED, which PD answers before it proposes a
+     *                     request it rejects (invalid, or a PD peer without the txn opcode), so
+     *                     nothing was applied. A call PD answered with a gRPC status carries
+     *                     that status in its message and as its cause (UNIMPLEMENTED: a PD
+     *                     from before this RPC rejected it before any apply; UNAVAILABLE,
+     *                     DEADLINE_EXCEEDED: the call was lost)
+     */
+    public TxnResponse txn(TxnRequest request) throws PDException {
+        TxnResponse response = blockingUnaryCallOnce(KvServiceGrpc.getTxnMethod(), request);
+        if (response == null) {
+            throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
+                                  "No response to a txn from PD, pd.peers=" +
+                                  config.getServerHost());
+        }
+        handleErrors(response.getHeader());
+        return response;
+    }
+
     private void onEvent(WatchResponse value, Consumer<T> consumer) {
         log.debug("receive message for {},event Count:{}", value, value.getEventsCount());
         if (value.getEventsCount() != 0) {
             try {
                 consumer.accept((T) value);
             } catch (Exception e) {
                 log.info(
                         "an error occurred while executing the client callback method, which " +