imbajin commented on code in PR #3376: URL: https://github.com/apache/hugegraph/pull/3376#discussion_r4327429752
########## 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: Blocking: yes, on a4784f957c41be9665cf6e8f65d2b171a7e7b3c6. The same-batch marker removal is present, but the new guarded cleanup still deletes a replacement value in a different interleaving: 1. `clearTTLData()` scans the old expired marker into `kvList`. 2. A TXN writes a different, non-TTL value and removes that marker atomically. 3. Cleanup reaches `getOne()` and reads the **new** value. 4. `removeExpired()` compares that new string, succeeds, and deletes it. The guard checks neither that the expired marker still exists nor that the current value belongs to that expiry. This does not require an identical-string rewrite. The new `testTtlCleanupKeepsAValueATxnWroteMeanwhile` covers replacement after reading the old value, but not replacement between the marker scan and `getOne()`. Please bind the deletion to the expired marker/value generation at the apply boundary and add this second interleaving as a regression test. Source: [KvService.java L337–345](https://github.com/apache/hugegraph/blob/a4784f957c41be9665cf6e8f65d2b171a7e7b3c6/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java#L337-L345), [L369–383](https://github.com/apache/hugegraph/blob/a4784f957c41be9665cf6e8f65d2b171a7e7b3c6/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java#L369-L383). Evidence: independently reproduced with the exact `clearTTLData`/`removeExpired` methods and a deterministic interleaving in a bounded Java harness. Storage, protobuf and transaction dependencies were stubbed; this was not a real RocksDB/JRaft cluster test. Update, 2026-10-20: the specific TXN interleaving above is fixed in e2ba916fdba53d2c6fc24219d528255cdf54b3d4. Cleanup now checks that the value's expiry matches the scanned marker and, at apply, compares both marker presence and value. This protects a TXN replacement before or after the value read, including an identical-string replacement. Independent bounded validation: five positive cases passed using the exact updated `clearTTLData`/`removeExpired` methods: replacement before the value read; replacement after it with a different string; replacement after it with the same string; a non-TTL replacement already read while its old marker remains; and normal deletion of an unchanged expired value. Two negative controls, removing the expiry guard or marker guard separately, failed as expected. This closes my reported TXN-cleanup finding, not the whole PR review. The harness used in-memory storage/protobuf/transaction stubs, not a real RocksDB/JRaft cluster. Ordinary-put/expiry-read concurrency remains outside this finding's fix, and no current-head CI success or full-cluster acceptance is claimed. -- 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]
