diff --git a/hugegraph-pd/docs/api-reference.md b/hugegraph-pd/docs/api-reference.md index 35d70fadd4..8cf592c7e7 100644 --- a/hugegraph-pd/docs/api-reference.md +++ b/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 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":,"inc":,"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; diff --git a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java index 44ca5dccee..58d3a6d4b0 100644 --- a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java +++ b/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; @@ -279,6 +281,88 @@ protected RespT blockingUnaryCall( 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 RespT blockingUnaryCallOnce( + MethodDescriptor method, ReqT req) throws PDException { + return blockingUnaryCallOnce(method, req, false); + } + + private RespT blockingUnaryCallOnce( + MethodDescriptor 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(hosts.length); private AbstractBlockingStub getConcurrentBlockingStub(String address) { AbstractBlockingStub stub = stubs.get(address); diff --git a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java index e513fa62e7..7480091352 100644 --- a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java +++ b/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,6 +193,33 @@ public TTLResponse putTTL(String key, String value, long ttl) throws PDException 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 consumer) { log.debug("receive message for {},event Count:{}", value, value.getEventsCount()); if (value.getEventsCount() != 0) { diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java index abb71cc8ec..4bc4dd2262 100644 --- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java @@ -18,6 +18,8 @@ package org.apache.hugegraph.pd; import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedList; import java.util.List; @@ -26,10 +28,13 @@ import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.config.PDConfig; import org.apache.hugegraph.pd.grpc.kv.Kv; +import org.apache.hugegraph.pd.grpc.kv.TxnRequest; +import org.apache.hugegraph.pd.grpc.kv.TxnResponse; import org.apache.hugegraph.pd.grpc.kv.V; import org.apache.hugegraph.pd.meta.MetadataKeyHelper; import org.apache.hugegraph.pd.meta.MetadataRocksDBStore; import org.apache.hugegraph.pd.store.KV; +import org.apache.hugegraph.pd.store.KvTxnApplier; import org.springframework.stereotype.Service; import com.google.protobuf.InvalidProtocolBufferException; @@ -71,6 +76,40 @@ public static byte[] getKeyBytes(Object... keys) { return key.getBytes(Charset.defaultCharset()); } + /** + * The TTL marker {@link #getTTLStoreKey(String, long)} wrote next to the store key of a + * value that expires at expireAt (the ttl field of its {@link V}), so every raft node + * names the same marker, the one getTTLStoreKey wrote. The TTL cleanup is meant to delete + * the key a marker names once the marker's time has passed, whatever the key holds by then + * (it deletes another name today, apache/hugegraph#3378), so a write that replaces a TTL + * value deletes its marker too. + */ + public static byte[] ttlMarkerKey(byte[] storeKey, long expireAt, String charsetName) { + /* + * K@T@