imbajin commented on code in PR #3376:
URL: https://github.com/apache/hugegraph/pull/3376#discussion_r4325378261
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/RaftKVStore.java:
##########
@@ -329,6 +393,25 @@ private boolean isLeader() {
@Override
public boolean invoke(KVOperation op, KVStoreClosure response) throws
PDException {
+ if (this.appliedBefore(op)) {
+ if (response != null) {
+ /*
+ * Not a replay: this node proposed the entry in this run, and
the log gave it
+ * an index the store claims to hold already. The store and
the log do not
+ * match (a store restored without its log, or one written on
the path
+ * without raft by an older release); refused, since a skipped
write would
+ * answer success and write nothing.
+ */
+ long last = KvTxnApplier.lastIndex(this.store);
+ String message = "Refused raft entry " + op.getIndex() + ":
the store holds " +
+ "TXN index " + last + " already, so the store
and the raft " +
+ "log do not match";
+ log.error(message);
+
response.setError(Pdpb.Error.newBuilder().setType(Pdpb.ErrorType.UNKNOWN)
+ .setMessage(message).build());
+ }
+ return false;
Review Comment:
‼️ Critical: Blocking: yes. Summary: This store/log mismatch path skips a
committed Raft entry on this node but does not fail the state-machine apply.
Evidence: RaftStateMachine.onApply ignores this handler return and still
completes Status.OK; in the documented store-ahead-of-log recovery case, peers
can apply the entry while this node stays stale. Fence this node and require
snapshot recovery before it serves reads or participates again.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/KVOperation.java:
##########
@@ -49,6 +49,11 @@ public class KVOperation {
public static final byte CLEAR = 0x07;
public static final byte PUT_WITH_TTL_UNIT = 0x08;
public static final byte REMOVE_WITH_TTL = 0x09;
+ /**
+ * Atomic multi-key write, the value holds a serialized kv.TxnEntry: the
request with its
+ * keys encoded by the proposing node
+ */
+ public static final byte TXN = 0x0A;
Review Comment:
‼️ Critical: Blocking: yes. Summary: This adds Raft opcode 0x0A, but older
PD peers only log an unknown op and return normally from invoke. Evidence:
during a rolling upgrade, an updated leader can commit TXN while old peers
advance the committed log without applying it; a later old leader can serve
stale metadata. Gate TXN proposals until every peer supports the opcode, or
make unsupported peers fail and recover before continuing.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/KvTxnApplier.java:
##########
@@ -0,0 +1,418 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.pd.store;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hugegraph.pd.KvService;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.Pdpb;
+import org.apache.hugegraph.pd.grpc.kv.TxnCompare;
+import org.apache.hugegraph.pd.grpc.kv.TxnEntry;
+import org.apache.hugegraph.pd.grpc.kv.TxnOp;
+import org.apache.hugegraph.pd.grpc.kv.TxnRecord;
+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 com.google.common.primitives.Bytes;
+import com.google.protobuf.ByteString;
+import com.google.gson.Gson;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import com.google.gson.JsonPrimitive;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Applies a KV TXN to a local store. Every raft node runs it on the apply
thread with the
+ * entry's log index, so it must be deterministic: it reads only the store and
the entry,
+ * never the clock (TTL expiry is ignored), and a failed compare or a rejected
record is a
+ * result, not an exception.
+ * <p>
+ * The node that proposes a TXN turns the request into its {@link TxnEntry}
with
+ * {@link #entry(TxnRequest)}: every key encoded as {@link
KvService#put(String, String)}
+ * encodes keys on that node, so KvService get and scan read what a TXN
writes, and every node
+ * writes the bytes the proposer chose, whatever its own default charset.
Values are wrapped
+ * as put wraps them.
+ * <p>
+ * Replay: jraft applies the entries after the last snapshot again on restart,
and without a
+ * snapshot that is over the store that holds them. {@link #LAST_INDEX_KEY} is
written with
+ * every TXN and holds its index, and RaftKVStore skips every store write (TXN
or plain) at or
+ * below it: all of them were applied before it. Skipping only the TXNs would
not do, since a
+ * plain put replayed before a skipped TXN, such as a schema write from before
the upgrade,
+ * would overwrite what the TXN wrote. Plain writes after the last TXN are
blind writes and
+ * replay in order to the same store. A store that runs without raft has no
log index and
+ * counts its TXNs under {@link #LOCAL_INDEX_KEY} instead, so that a store
later run with raft
+ * does not take that count for a raft index.
+ * <p>
+ * A TXN is not deduplicated: a TXN sent again applies again, and a second
apply is not always
+ * harmless (a guarded CREATE fails its record compare the second time). So
KvClient sends a
+ * TXN once, and a leader whose apply failed answers with the error instead of
forwarding it.
+ */
+@Slf4j
+public final class KvTxnApplier {
+
+ public static final String LIVE = "LIVE";
+ public static final String DROPPED = "DROPPED";
+ // The origin mark of a record a CREATE wrote, "src":"create", see
GraphRecord
+ public static final String CREATED = "create";
+
+ /**
+ * Raw store key outside the K@ space of KvService, so no KV call reads or
writes it: the
+ * raft index of the last TXN applied through raft
+ */
+ public static final byte[] LAST_INDEX_KEY =
+ "__TXN_LAST_INDEX".getBytes(StandardCharsets.UTF_8);
+
+ /**
+ * The TXN count of a store that runs without raft (raft.enable=false),
which stands in
+ * for the raft index there; never read on the raft path
+ */
+ public static final byte[] LOCAL_INDEX_KEY =
+ "__TXN_LOCAL_INDEX".getBytes(StandardCharsets.UTF_8);
+
+ private static final Gson GSON = new Gson();
+
+ /**
+ * Stands for a stored record that can't be read
+ */
+ private static final GraphRecord INVALID = new GraphRecord(0L, 0L, null,
null);
+
+ private static volatile TxnApplyListener listener;
+
+ private KvTxnApplier() {
+ }
+
+ /**
+ * The registration point of schema sync dispatch; null (the default)
calls nothing
+ */
+ public static void setListener(TxnApplyListener txnListener) {
+ listener = txnListener;
+ }
+
+ public static TxnApplyListener listener() {
+ return listener;
+ }
+
+ /**
+ * Rejects what the apply could not handle, before the TXN is proposed
+ */
+ public static void validate(TxnRequest request) throws PDException {
+ for (TxnCompare compare : request.getComparesList()) {
+ if (compare.getKey().isEmpty()) {
+ throw invalid("a TXN compare needs a key");
+ }
+ if (compare.getKind() == TxnCompare.Kind.UNRECOGNIZED) {
+ throw invalid("unknown compare kind on key " +
compare.getKey());
+ }
+ }
+ for (TxnOp op : request.getOpsList()) {
+ // An empty key is the whole KV space to a DELETE_PREFIX
+ if (op.getKey().isEmpty()) {
+ throw invalid("a TXN op needs a key");
+ }
+ if (op.getType() == TxnOp.Type.UNRECOGNIZED) {
+ throw invalid("unknown op type on key " + op.getKey());
+ }
+ }
+ if (!request.hasRecord()) {
+ return;
+ }
+ TxnRecord record = request.getRecord();
+ if (record.getKey().isEmpty() || record.getOp() ==
TxnRecord.Op.UNRECOGNIZED) {
+ throw invalid("a TXN record needs a key and a known op");
+ }
+ if (record.getOp() == TxnRecord.Op.DROP &&
record.getExpectedIncarnation() <= 0L) {
+ throw invalid("DROP needs the expected incarnation");
+ }
+ }
+
+ /**
+ * The raft index of the last TXN applied through raft, 0 before the first
+ */
+ public static long lastIndex(HgKVStore store) throws PDException {
+ return lastIndex(store, LAST_INDEX_KEY);
+ }
+
+ public static long lastIndex(HgKVStore store, byte[] indexKey) throws
PDException {
+ byte[] bytes = store.get(indexKey);
+ if (bytes == null || bytes.length == 0) {
+ return 0L;
+ }
+ return Long.parseLong(new String(bytes, StandardCharsets.UTF_8));
+ }
+
+ /**
+ * The entry of a request, with its keys encoded on this node: built once,
by the node
+ * that proposes the TXN, or by the store without raft that applies it
+ */
+ public static TxnEntry entry(TxnRequest request) {
+ TxnEntry.Builder entry = TxnEntry.newBuilder().setRequest(request);
+ for (TxnCompare compare : request.getComparesList()) {
+
entry.addKeys(ByteString.copyFrom(KvService.getKeyBytes(compare.getKey())));
+ }
+ for (TxnOp op : request.getOpsList()) {
+
entry.addKeys(ByteString.copyFrom(KvService.getKeyBytes(op.getKey())));
+ }
+ if (request.hasRecord()) {
+
entry.addKeys(ByteString.copyFrom(KvService.getKeyBytes(request.getRecord().getKey())));
+ }
+ return entry.build();
+ }
+
+ /**
+ * Applies an entry of the raft log at its index, with the registered
listener
+ */
+ public static TxnResponse apply(HgKVStore store, TxnEntry entry, long
index)
+ throws PDException {
+ return apply(store, entry, index, LAST_INDEX_KEY, listener);
+ }
+
+ /**
+ * @param indexKey the key that holds the index of the last applied TXN:
+ * {@link #LAST_INDEX_KEY} on the raft path, {@link
#LOCAL_INDEX_KEY} on a
+ * store without raft
+ */
+ public static TxnResponse apply(HgKVStore store, TxnEntry entry, long
index,
+ byte[] indexKey, TxnApplyListener
txnListener)
+ throws PDException {
+ if (lastIndex(store, indexKey) >= index) {
+ // The store holds this entry already, and no caller waits on a
replay
+ return TxnResponse.getDefaultInstance();
+ }
+ TxnRequest request = entry.getRequest();
+ int keys = request.getComparesCount() + request.getOpsCount() +
+ (request.hasRecord() ? 1 : 0);
+ if (entry.getKeysCount() != keys) {
+ throw invalid("a TXN entry needs one store key per compare, op and
record, has " +
+ entry.getKeysCount() + " for " + keys);
+ }
+ // Written whatever the outcome, so that a replay does not run the TXN
again
+ List<KV> batch = new ArrayList<>();
+ batch.add(new KV(indexKey,
Long.toString(index).getBytes(StandardCharsets.UTF_8)));
+ TxnResponse response = evaluate(store, entry, index, batch);
+ store.writeBatch(batch);
+ if (response.getSucceeded() && request.hasRecord()) {
+ // The listener takes rev, inc and state; the origin mark is not
passed on
+ notify(txnListener, index, request.getRecord().getKey(), new
GraphRecord(
+ index, response.getIncarnation(), response.getState(),
null));
+ }
+ return response;
+ }
+
+ /**
+ * Runs the compares and the record check, and adds the writes of a TXN
that passes them
+ * to the batch
+ */
+ private static TxnResponse evaluate(HgKVStore store, TxnEntry entry, long
index,
+ List<KV> batch) throws PDException {
+ TxnRequest request = entry.getRequest();
+ TxnResponse.Builder response = TxnResponse.newBuilder();
+ int next = 0;
+ for (int i = 0; i < request.getComparesCount(); i++) {
+ if (!matches(store, request.getCompares(i),
entry.getKeys(next++).toByteArray())) {
+ return response.setFailure(TxnResponse.Failure.COMPARE_FAILED)
+ .setFailedCompare(i).build();
+ }
+ }
+
+ byte[] recordKey = null;
+ GraphRecord record = null;
+ if (request.hasRecord()) {
+ recordKey = entry.getKeys(entry.getKeysCount() - 1).toByteArray();
+ GraphRecord stored = readRecord(store.get(recordKey));
+ TxnResponse.Failure failure = stored == INVALID ?
+ TxnResponse.Failure.INVALID_RECORD :
+ check(request.getRecord(), stored);
+ if (failure != TxnResponse.Failure.NONE) {
+ return response.setFailure(failure)
+ .setIncarnation(stored == null ? 0L :
stored.inc)
+ .setState(stored == null || stored.state ==
null ?
+ "" : stored.state).build();
+ }
+ record = next(request.getRecord().getOp(), stored, index);
+ }
+
+ int first = batch.size();
+ for (TxnOp op : request.getOpsList()) {
+ byte[] key = entry.getKeys(next++).toByteArray();
+ switch (op.getType()) {
+ case PUT:
+ batch.add(new KV(key, wrap(op.getValue())));
Review Comment:
⚠️ Important: Blocking: yes. Summary: This PUT writes a zero-TTL wrapper but
leaves any prior TTL index marker in the store. Evidence: KvService.put(key,
value, ttl) records a T@...@key marker; clearTTLData later deletes that key
when the marker expires, so it can delete this newer transaction value. Remove
or reconcile stale TTL markers in the batch, or reject TTL-managed keys
explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
