From 39d0f5b30f987f0cd73eb5ce2ad9294319aadaea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=8C=E4=B8=AB=E8=AE=B2=E6=A2=B5?= Date: Mon, 27 Jul 2026 13:18:53 +0800 Subject: [PATCH] fix(etcd): preserve raw key identity in browser actions --- .github/workflows/ci.yml | 1 + .../java/com/dbx/agent/AgentProtocol.java | 18 +- .../src/main/resources/agent-protocol-v1.json | 62 +- .../src/main/resources/agent-protocol-v2.json | 65 +- .../java/com/dbx/agent/etcd/EtcdAgent.java | 553 ++++++- .../com/dbx/agent/etcd/EtcdAgentTest.java | 69 + .../src/components/etcd/EtcdDashboard.vue | 584 +++++++ .../src/components/etcd/EtcdKeyBrowser.vue | 659 +++++++- .../EtcdDashboardConcurrency.spec.ts | 15 + .../etcd/__tests__/EtcdKeyBrowser.spec.ts | 156 +- .../src/components/kv/KvKeyBrowser.vue | 1067 ++++++++++--- .../src/components/kv/KvValueEditor.vue | 102 ++ .../kv/__tests__/KvKeyBrowserExport.spec.ts | 42 + .../kv/__tests__/KvKeyBrowserLayout.spec.ts | 23 + .../KvKeyBrowserMutationError.spec.ts | 17 + .../KvKeyBrowserSearchAndSplit.spec.ts | 20 + .../kv/__tests__/KvKeyBrowserTtl.spec.ts | 19 + .../src/components/layout/AppTabBar.vue | 3 + .../src/components/layout/ContentArea.vue | 10 + .../src/components/sidebar/ConnectionTree.vue | 2 +- .../sidebar/SidebarTreeRuntimeHost.vue | 6 +- .../src/components/sidebar/TreeItem.vue | 3 + apps/desktop/src/i18n/locales/en.ts | 147 +- apps/desktop/src/i18n/locales/es.ts | 145 +- apps/desktop/src/i18n/locales/it.ts | 147 +- apps/desktop/src/i18n/locales/ja.ts | 145 +- apps/desktop/src/i18n/locales/pt-BR.ts | 147 +- apps/desktop/src/i18n/locales/zh-CN.ts | 147 +- apps/desktop/src/i18n/locales/zh-TW.ts | 145 +- .../__tests__/kv/etcdDashboardError.spec.ts | 23 + .../__tests__/kv/etcdDashboardMetrics.spec.ts | 74 + .../lib/__tests__/kv/kvExportScope.spec.ts | 31 + .../src/lib/__tests__/kv/kvKeySearch.spec.ts | 15 + .../src/lib/__tests__/kv/kvKeyTree.spec.ts | 38 +- .../__tests__/kv/kvMetadataRefresh.spec.ts | 34 +- .../lib/__tests__/kv/kvMutationError.spec.ts | 28 + .../__tests__/kv/kvRefreshSelection.spec.ts | 13 +- .../src/lib/__tests__/kv/kvTtl.spec.ts | 17 +- .../lib/__tests__/kv/kvValueFormat.spec.ts | 40 + .../sidebar/sidebarTreeItemLayout.spec.ts | 7 +- apps/desktop/src/lib/backend/api.ts | 13 +- apps/desktop/src/lib/backend/http.ts | 29 +- apps/desktop/src/lib/backend/tauri.ts | 186 ++- apps/desktop/src/lib/kv/etcdDashboardError.ts | 28 + .../src/lib/kv/etcdDashboardMetrics.ts | 95 ++ apps/desktop/src/lib/kv/kvExportScope.ts | 36 + apps/desktop/src/lib/kv/kvKeySearch.ts | 14 + apps/desktop/src/lib/kv/kvKeyTree.ts | 86 +- apps/desktop/src/lib/kv/kvMetadataRefresh.ts | 23 +- apps/desktop/src/lib/kv/kvMutationError.ts | 40 + apps/desktop/src/lib/kv/kvRefreshSelection.ts | 6 +- apps/desktop/src/lib/kv/kvTtl.ts | 13 + apps/desktop/src/lib/kv/kvValueDisplay.ts | 2 +- apps/desktop/src/lib/kv/kvValueFormat.ts | 102 ++ .../src/lib/sidebar/sidebarActiveTabTarget.ts | 12 + .../src/lib/sidebar/sidebarTreeItemLayout.ts | 2 + apps/desktop/src/lib/sidebar/treeNodeClick.ts | 2 +- apps/desktop/src/lib/sidebar/treeNodeIcon.ts | 4 +- apps/desktop/src/lib/tabs/tabPresentation.ts | 5 + apps/desktop/src/stores/connectionStore.ts | 9 + apps/desktop/src/types/database.ts | 3 +- crates/dbx-core/assets/agent-protocol-v1.json | 62 +- crates/dbx-core/assets/agent-protocol-v2.json | 65 +- crates/dbx-core/src/agent_kv.rs | 1368 ++++++++++++++++- crates/dbx-core/src/db/agent_driver.rs | 82 +- crates/dbx-web/src/main.rs | 3 + crates/dbx-web/src/routes/etcd.rs | 100 +- src-tauri/src/commands/etcd_cmd.rs | 87 +- src-tauri/src/lib.rs | 3 + 69 files changed, 6577 insertions(+), 742 deletions(-) create mode 100644 apps/desktop/src/components/etcd/EtcdDashboard.vue create mode 100644 apps/desktop/src/components/etcd/__tests__/EtcdDashboardConcurrency.spec.ts create mode 100644 apps/desktop/src/components/kv/KvValueEditor.vue create mode 100644 apps/desktop/src/components/kv/__tests__/KvKeyBrowserExport.spec.ts create mode 100644 apps/desktop/src/components/kv/__tests__/KvKeyBrowserLayout.spec.ts create mode 100644 apps/desktop/src/components/kv/__tests__/KvKeyBrowserMutationError.spec.ts create mode 100644 apps/desktop/src/components/kv/__tests__/KvKeyBrowserSearchAndSplit.spec.ts create mode 100644 apps/desktop/src/components/kv/__tests__/KvKeyBrowserTtl.spec.ts create mode 100644 apps/desktop/src/lib/__tests__/kv/etcdDashboardError.spec.ts create mode 100644 apps/desktop/src/lib/__tests__/kv/etcdDashboardMetrics.spec.ts create mode 100644 apps/desktop/src/lib/__tests__/kv/kvExportScope.spec.ts create mode 100644 apps/desktop/src/lib/__tests__/kv/kvKeySearch.spec.ts create mode 100644 apps/desktop/src/lib/__tests__/kv/kvMutationError.spec.ts create mode 100644 apps/desktop/src/lib/__tests__/kv/kvValueFormat.spec.ts create mode 100644 apps/desktop/src/lib/kv/etcdDashboardError.ts create mode 100644 apps/desktop/src/lib/kv/etcdDashboardMetrics.ts create mode 100644 apps/desktop/src/lib/kv/kvExportScope.ts create mode 100644 apps/desktop/src/lib/kv/kvKeySearch.ts create mode 100644 apps/desktop/src/lib/kv/kvMutationError.ts create mode 100644 apps/desktop/src/lib/kv/kvValueFormat.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6643a72b24..4a0913db22 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -311,6 +311,7 @@ jobs: filters: | frontend: - 'apps/desktop/**' + - 'docs/**' - 'packages/**' - 'pnpm-lock.yaml' - 'package.json' diff --git a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java index 9b7ce35fe2..868050d89e 100644 --- a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java +++ b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java @@ -67,6 +67,9 @@ public final class AgentProtocol { public static final String KV_METHOD_GET = "kv_get"; public static final String KV_METHOD_PUT = "kv_put"; public static final String KV_METHOD_DELETE = "kv_delete"; + public static final String KV_METHOD_RENAME = "kv_rename"; + public static final String KV_METHOD_HISTORY = "kv_history"; + public static final String KV_METHOD_STATUS = "kv_status"; public static final String CAPABILITY_CONNECT = "connect"; public static final String CAPABILITY_TEST_CONNECTION = "test_connection"; @@ -77,6 +80,10 @@ public final class AgentProtocol { public static final String CAPABILITY_DDL = "ddl"; public static final String CAPABILITY_KV = "kv"; public static final String CAPABILITY_KV_TTL = "kv_ttl"; + public static final String CAPABILITY_KV_CAS = "kv_cas"; + public static final String CAPABILITY_KV_LIST_VALUES = "kv_list_values"; + public static final String CAPABILITY_KV_STATUS = "kv_status"; + public static final String CAPABILITY_KV_HISTORY = "kv_history"; public static final String CAPABILITY_MULTI_SESSION = "multi_session"; public static final List CAPABILITIES = Collections.unmodifiableList(Arrays.asList( @@ -98,7 +105,11 @@ public final class AgentProtocol { CAPABILITY_TRANSACTION, CAPABILITY_DDL, CAPABILITY_KV, - CAPABILITY_KV_TTL + CAPABILITY_KV_TTL, + CAPABILITY_KV_CAS, + CAPABILITY_KV_LIST_VALUES, + CAPABILITY_KV_STATUS, + CAPABILITY_KV_HISTORY )); public static final List COMMON_METHODS = Collections.unmodifiableList(Arrays.asList( @@ -171,7 +182,10 @@ public final class AgentProtocol { KV_METHOD_LIST_PREFIX, KV_METHOD_GET, KV_METHOD_PUT, - KV_METHOD_DELETE + KV_METHOD_DELETE, + KV_METHOD_RENAME, + KV_METHOD_HISTORY, + KV_METHOD_STATUS )); private AgentProtocol() { diff --git a/agents/common/src/main/resources/agent-protocol-v1.json b/agents/common/src/main/resources/agent-protocol-v1.json index 63cd20bf50..cbc33d0f4c 100644 --- a/agents/common/src/main/resources/agent-protocol-v1.json +++ b/agents/common/src/main/resources/agent-protocol-v1.json @@ -1,40 +1,10 @@ { "protocolVersion": 1, "handshakeMethod": "handshake", - "handshakeResponseFields": [ - "protocolVersion", - "agentProtocolVersion", - "capabilities" - ], - "allCapabilities": [ - "connect", - "test_connection", - "metadata", - "query", - "paged_query", - "transaction", - "ddl", - "kv", - "kv_ttl" - ], - "capabilities": [ - "connect", - "test_connection", - "metadata", - "query", - "paged_query", - "transaction", - "ddl" - ], - "defaultSqlCapabilities": [ - "connect", - "test_connection", - "metadata", - "query", - "paged_query", - "transaction", - "ddl" - ], + "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history"], + "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], + "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "commonMethods": [ "handshake", "connect", @@ -69,26 +39,6 @@ "disconnect", "shutdown" ], - "mongoLegacyMethods": [ - "list_databases", - "list_collections", - "find_documents", - "find_documents_extended_json", - "count_documents", - "server_version", - "create_index", - "drop_indexes", - "drop_collection", - "insert_document", - "update_document", - "update_documents", - "delete_document", - "delete_documents" - ], - "kvMethods": [ - "kv_list_prefix", - "kv_get", - "kv_put", - "kv_delete" - ] + "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "drop_indexes", "drop_collection", "insert_document", "update_document", "update_documents", "delete_document", "delete_documents"], + "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status"] } diff --git a/agents/common/src/main/resources/agent-protocol-v2.json b/agents/common/src/main/resources/agent-protocol-v2.json index a00b78a1c6..33fbd99a0b 100644 --- a/agents/common/src/main/resources/agent-protocol-v2.json +++ b/agents/common/src/main/resources/agent-protocol-v2.json @@ -1,43 +1,10 @@ { "protocolVersion": 2, "handshakeMethod": "handshake", - "handshakeResponseFields": [ - "protocolVersion", - "agentProtocolVersion", - "capabilities" - ], - "allCapabilities": [ - "connect", - "test_connection", - "metadata", - "query", - "paged_query", - "transaction", - "ddl", - "kv", - "kv_ttl", - "multi_session" - ], - "capabilities": [ - "connect", - "test_connection", - "metadata", - "query", - "paged_query", - "transaction", - "ddl", - "multi_session" - ], - "defaultSqlCapabilities": [ - "connect", - "test_connection", - "metadata", - "query", - "paged_query", - "transaction", - "ddl", - "multi_session" - ], + "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "multi_session"], + "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"], + "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"], "commonMethods": [ "handshake", "connect", @@ -76,28 +43,8 @@ "disconnect", "shutdown" ], - "mongoLegacyMethods": [ - "list_databases", - "list_collections", - "find_documents", - "find_documents_extended_json", - "count_documents", - "server_version", - "create_index", - "drop_indexes", - "drop_collection", - "insert_document", - "update_document", - "update_documents", - "delete_document", - "delete_documents" - ], - "kvMethods": [ - "kv_list_prefix", - "kv_get", - "kv_put", - "kv_delete" - ], + "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "drop_indexes", "drop_collection", "insert_document", "update_document", "update_documents", "delete_document", "delete_documents"], + "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status"], "sessionField": "agentSessionId", "cursorSessionField": "sessionId" } diff --git a/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java b/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java index 9c33cede80..fb596d3fe6 100644 --- a/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java +++ b/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java @@ -9,20 +9,33 @@ import io.grpc.netty.GrpcSslContexts; import io.etcd.jetcd.ByteSequence; import io.etcd.jetcd.Client; import io.etcd.jetcd.ClientBuilder; +import io.etcd.jetcd.Cluster; import io.etcd.jetcd.KV; import io.etcd.jetcd.KeyValue; -import io.etcd.jetcd.kv.DeleteResponse; -import io.etcd.jetcd.kv.GetResponse; -import io.etcd.jetcd.kv.PutResponse; +import io.etcd.jetcd.Lease; +import io.etcd.jetcd.Maintenance; +import io.etcd.jetcd.Watch; +import io.etcd.jetcd.cluster.Member; +import io.etcd.jetcd.cluster.MemberListResponse; import io.etcd.jetcd.kv.TxnResponse; import io.etcd.jetcd.lease.LeaseGrantResponse; import io.etcd.jetcd.lease.LeaseTimeToLiveResponse; +import io.etcd.jetcd.maintenance.AlarmMember; +import io.etcd.jetcd.maintenance.AlarmResponse; +import io.etcd.jetcd.maintenance.StatusResponse; import io.etcd.jetcd.op.Cmp; import io.etcd.jetcd.op.CmpTarget; import io.etcd.jetcd.op.Op; +import io.etcd.jetcd.kv.DeleteResponse; +import io.etcd.jetcd.kv.GetResponse; +import io.etcd.jetcd.kv.PutResponse; +import io.etcd.jetcd.options.DeleteOption; import io.etcd.jetcd.options.GetOption; import io.etcd.jetcd.options.LeaseOption; import io.etcd.jetcd.options.PutOption; +import io.etcd.jetcd.options.WatchOption; +import io.etcd.jetcd.watch.WatchEvent; +import io.etcd.jetcd.watch.WatchResponse; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; import java.io.BufferedReader; @@ -34,26 +47,43 @@ import java.nio.charset.CharsetDecoder; import java.nio.charset.CodingErrorAction; import java.util.Arrays; import java.util.ArrayList; +import java.util.ArrayDeque; import java.util.Base64; import java.util.Collections; +import java.util.Deque; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; public final class EtcdAgent { private static final Gson GSON = new Gson(); private static final int DEFAULT_LIMIT = 100; private static final int RPC_TIMEOUT_SECONDS = 30; private static final int PRESERVE_LEASE_MAX_ATTEMPTS = 3; + private static final long HISTORY_DEFAULT_REVISION_WINDOW = 10_000L; private static final List CAPABILITIES = Collections.unmodifiableList(Arrays.asList( AgentProtocol.CAPABILITY_CONNECT, AgentProtocol.CAPABILITY_TEST_CONNECTION, AgentProtocol.CAPABILITY_KV, - AgentProtocol.CAPABILITY_KV_TTL + AgentProtocol.CAPABILITY_KV_TTL, + AgentProtocol.CAPABILITY_KV_CAS, + AgentProtocol.CAPABILITY_KV_LIST_VALUES, + AgentProtocol.CAPABILITY_KV_STATUS, + AgentProtocol.CAPABILITY_KV_HISTORY )); private static Client client; private static KV kv; + private static List connectedEndpoints = Collections.emptyList(); private EtcdAgent() { } @@ -69,6 +99,7 @@ public final class EtcdAgent { closeClient(); client = nextClient; kv = client.getKVClient(); + connectedEndpoints = endpoints(connection); return Collections.singletonMap("ok", true); } @@ -141,16 +172,21 @@ public final class EtcdAgent { KV active = requireKv(); String prefix = stringOrDefault(params, "prefix", ""); int limit = intOrDefault(params, "limit", DEFAULT_LIMIT); + Long revision = longOrNull(params, "revision"); + boolean includeValues = boolOrDefault(params, "includeValues", false); String continuation = stringOrNull(params, "continuation"); ByteSequence start = continuation == null || continuation.isBlank() ? prefixStart(prefix) : ByteSequence.from(Base64.getDecoder().decode(continuation)); - GetOption option = GetOption.newBuilder() + GetOption.Builder optionBuilder = GetOption.newBuilder() .withRange(prefixEnd(byteSequence(prefix))) .withLimit(Math.max(1, limit)) .withSortField(GetOption.SortTarget.KEY) - .withSortOrder(GetOption.SortOrder.ASCEND) - .build(); + .withSortOrder(GetOption.SortOrder.ASCEND); + if (revision != null && revision > 0) { + optionBuilder.withRevision(revision); + } + GetOption option = optionBuilder.build(); GetResponse response = active.get(start, option).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); List> keys = new ArrayList<>(); @@ -158,24 +194,29 @@ public final class EtcdAgent { for (KeyValue item : kvs) { Map row = metadata(item); row.put("key", displayBytes(item.getKey().getBytes())); + row.put("keyBytes", bytesObject(item.getKey().getBytes())); + if (includeValues) { + row.put("value", valueObject(item.getValue().getBytes())); + } keys.add(row); } Map result = new LinkedHashMap<>(); result.put("keys", keys); result.put("continuation", response.isMore() && !kvs.isEmpty() ? nextContinuation(kvs.get(kvs.size() - 1)) : null); - result.put("revision", response.getHeader().getRevision()); + result.put("revision", longString(response.getHeader().getRevision())); return result; } private static Object get(JsonObject params) throws Exception { KV active = requireKv(); - String key = params.get("key").getAsString(); - boolean metadataOnly = boolOrDefault(params, "metadataOnly", false); - GetOption option = metadataOnly - ? GetOption.builder().withKeysOnly(true).build() - : GetOption.DEFAULT; - GetResponse response = active.get(byteSequence(key), option).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + ByteSequence key = keyBytes(params); + Long revision = longOrNull(params, "revision"); + GetOption.Builder optionBuilder = GetOption.newBuilder(); + if (revision != null && revision > 0) { + optionBuilder.withRevision(revision); + } + GetResponse response = active.get(key, optionBuilder.build()).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); Map result = new LinkedHashMap<>(); if (response.getKvs().isEmpty()) { result.put("found", false); @@ -185,27 +226,24 @@ public final class EtcdAgent { return result; } KeyValue item = response.getKvs().get(0); - Map metadata = metadata(item); - if (item.getLease() > 0) { - LeaseTimeToLiveResponse lease = - requireClient().getLeaseClient().timeToLive(item.getLease(), LeaseOption.DEFAULT) - .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); - metadata.put("ttl", lease.getTTL()); - } result.put("found", true); result.put("key", displayBytes(item.getKey().getBytes())); - result.put("value", metadataOnly ? null : valueObject(item.getValue().getBytes())); - result.put("metadata", metadata); + result.put("keyBytes", bytesObject(item.getKey().getBytes())); + if (!boolOrDefault(params, "metadataOnly", false)) { + result.put("value", valueObject(item.getValue().getBytes())); + } else { + result.put("value", null); + } + result.put("metadata", metadataWithTtl(item)); return result; } private static Object put(JsonObject params) throws Exception { KV active = requireKv(); - Client activeClient = requireClient(); - String key = params.get("key").getAsString(); + ByteSequence key = keyBytes(params); byte[] value = parseValue(params.getAsJsonObject("value")); - ByteSequence keyBytes = byteSequence(key); - ByteSequence valueBytes = ByteSequence.from(value); + Long expectedModRevision = longOrNull(params, "expectedModRevision"); + Long expectedCreateRevision = longOrNull(params, "expectedCreateRevision"); JsonElement leaseElement = params.get("lease"); JsonElement ttlElement = params.get("ttl"); boolean hasLease = leaseElement != null && !leaseElement.isJsonNull(); @@ -214,43 +252,56 @@ public final class EtcdAgent { if ((hasLease && hasTtl) || (preserveLease && (hasLease || hasTtl))) { throw new IllegalArgumentException("lease, ttl, and preserveLease cannot be specified together"); } - if (preserveLease) { - long revision = putPreservingLease(active, keyBytes, valueBytes); - return Collections.singletonMap("revision", revision); + long revision = putPreservingLease(active, key, ByteSequence.from(value)); + return Collections.singletonMap("revision", longString(revision)); } - - Long leaseId = null; + PutOption option = PutOption.DEFAULT; long grantedLeaseId = 0; if (hasTtl) { long ttl = ttlElement.getAsLong(); - if (ttl <= 0) { - throw new IllegalArgumentException("ttl must be a positive integer"); - } - LeaseGrantResponse grant = - activeClient.getLeaseClient().grant(ttl).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (ttl <= 0) throw new IllegalArgumentException("ttl must be a positive integer"); + LeaseGrantResponse grant = requireClient().getLeaseClient().grant(ttl).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); grantedLeaseId = grant.getID(); - leaseId = grantedLeaseId; + option = PutOption.newBuilder().withLeaseId(grantedLeaseId).build(); } else if (hasLease) { - leaseId = leaseElement.getAsLong(); + option = PutOption.newBuilder().withLeaseId(leaseElement.getAsLong()).build(); } - - try { - PutResponse response; - if (leaseId != null) { - PutOption option = PutOption.newBuilder().withLeaseId(leaseId).build(); - response = active.put(keyBytes, valueBytes, option) - .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); - } else { - response = active.put(keyBytes, valueBytes) + final long leaseToCleanUp = grantedLeaseId; + final Lease leaseClient = leaseToCleanUp == 0 ? null : requireClient().getLeaseClient(); + final PutOption writeOption = option; + return cleanUpGrantedLeaseOnFailure(leaseToCleanUp, leaseClient, () -> { + if (expectedModRevision != null || expectedCreateRevision != null) { + List comparisons = new ArrayList<>(); + if (expectedModRevision != null) { + comparisons.add(new Cmp(key, Cmp.Op.EQUAL, CmpTarget.modRevision(expectedModRevision))); + } + if (expectedCreateRevision != null) { + comparisons.add(new Cmp(key, Cmp.Op.EQUAL, CmpTarget.createRevision(expectedCreateRevision))); + } + TxnResponse response = active.txn() + .If(comparisons.toArray(Cmp[]::new)) + .Then(Op.put(key, ByteSequence.from(value), writeOption)) + .commit() .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (!response.isSucceeded()) { + throw new IllegalStateException("ETCD_CAS_CONFLICT: key changed after it was loaded"); + } + return Collections.singletonMap("revision", longString(response.getHeader().getRevision())); } - return Collections.singletonMap("revision", response.getHeader().getRevision()); + PutResponse response = active.put(key, ByteSequence.from(value), writeOption) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Collections.singletonMap("revision", longString(response.getHeader().getRevision())); + }); + } + + static T cleanUpGrantedLeaseOnFailure(long grantedLeaseId, Lease leaseClient, Callable write) throws Exception { + try { + return write.call(); } catch (Exception error) { if (grantedLeaseId != 0) { try { - activeClient.getLeaseClient().revoke(grantedLeaseId) - .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + leaseClient.revoke(grantedLeaseId).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (Exception revokeError) { error.addSuppressed(revokeError); } @@ -261,8 +312,7 @@ public final class EtcdAgent { static long putPreservingLease(KV active, ByteSequence key, ByteSequence value) throws Exception { for (int attempt = 0; attempt < PRESERVE_LEASE_MAX_ATTEMPTS; attempt++) { - GetResponse existing = - active.get(key, GetOption.DEFAULT).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + GetResponse existing = active.get(key, GetOption.DEFAULT).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); if (existing.getKvs().isEmpty() || existing.getKvs().get(0).getLease() <= 0) { throw new IllegalStateException("Cannot preserve lease: key does not exist or has no lease"); } @@ -284,11 +334,330 @@ public final class EtcdAgent { private static Object delete(JsonObject params) throws Exception { KV active = requireKv(); - DeleteResponse response = - active.delete(byteSequence(params.get("key").getAsString())).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + ByteSequence key = keyBytes(params); + Long expectedModRevision = longOrNull(params, "expectedModRevision"); + if (expectedModRevision != null) { + TxnResponse txnResponse = active.txn() + .If(new Cmp(key, Cmp.Op.EQUAL, CmpTarget.modRevision(expectedModRevision))) + .Then(Op.delete(key, DeleteOption.DEFAULT)) + .commit() + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (!txnResponse.isSucceeded()) { + throw new IllegalStateException("ETCD_CAS_CONFLICT: key changed after it was loaded"); + } + Map result = new LinkedHashMap<>(); + result.put("deleted", 1); + result.put("revision", longString(txnResponse.getHeader().getRevision())); + return result; + } + DeleteResponse response = active.delete(key).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); Map result = new LinkedHashMap<>(); result.put("deleted", response.getDeleted()); - result.put("revision", response.getHeader().getRevision()); + result.put("revision", longString(response.getHeader().getRevision())); + return result; + } + + private static Object rename(JsonObject params) throws Exception { + KV active = requireKv(); + ByteSequence sourceKey = keyBytes(params); + ByteSequence targetKey = byteSequence(params.get("newKey").getAsString()); + if (Arrays.equals(sourceKey.getBytes(), targetKey.getBytes())) { + Map unchanged = new LinkedHashMap<>(); + unchanged.put("renamed", true); + unchanged.put("revision", null); + return unchanged; + } + + GetResponse sourceResponse = active.get(sourceKey).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (sourceResponse.getKvs().isEmpty()) { + throw new IllegalStateException("ETCD_NOT_FOUND: source key does not exist"); + } + KeyValue source = sourceResponse.getKvs().get(0); + Long expected = longOrNull(params, "expectedModRevision"); + long expectedRevision = expected == null ? source.getModRevision() : expected; + PutOption putOption = source.getLease() == 0 + ? PutOption.DEFAULT + : PutOption.newBuilder().withLeaseId(source.getLease()).build(); + + TxnResponse response = active.txn() + .If( + new Cmp(sourceKey, Cmp.Op.EQUAL, CmpTarget.modRevision(expectedRevision)), + new Cmp(targetKey, Cmp.Op.EQUAL, CmpTarget.createRevision(0)) + ) + .Then( + Op.put(targetKey, source.getValue(), putOption), + Op.delete(sourceKey, DeleteOption.DEFAULT) + ) + .commit() + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (!response.isSucceeded()) { + throw new IllegalStateException("ETCD_CAS_CONFLICT: source changed or target already exists"); + } + Map result = new LinkedHashMap<>(); + result.put("renamed", true); + result.put("revision", longString(response.getHeader().getRevision())); + return result; + } + + private static Object history(JsonObject params) throws Exception { + requireKv(); + ByteSequence key = keyBytes(params); + int limit = Math.min(Math.max(1, intOrDefault(params, "limit", 100)), 500); + Long requestedEnd = longOrNull(params, "endRevision"); + GetOption.Builder latestOption = GetOption.newBuilder(); + if (requestedEnd != null && requestedEnd > 0) { + latestOption.withRevision(requestedEnd); + } + GetResponse latest = kv.get(key, latestOption.build()).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + long endRevision = requestedEnd == null ? latest.getHeader().getRevision() : requestedEnd; + long targetKeyRevision = latest.getKvs().isEmpty() + ? endRevision + : latest.getKvs().get(0).getModRevision(); + Long requestedStart = longOrNull(params, "startRevision"); + long startRevision = historyStartRevision(requestedStart, targetKeyRevision); + if (startRevision > targetKeyRevision) { + return Map.of( + "events", Collections.emptyList(), + "observedRevision", longString(endRevision), + "truncated", false + ); + } + + Deque> events = new ArrayDeque<>(limit); + AtomicBoolean truncated = new AtomicBoolean(requestedStart == null && startRevision > 1L); + AtomicReference failure = new AtomicReference<>(); + CountDownLatch created = new CountDownLatch(1); + CountDownLatch completed = new CountDownLatch(1); + Watch watch = client.getWatchClient(); + WatchOption option = WatchOption.newBuilder() + .withRevision(startRevision) + .withPrevKV(true) + .withProgressNotify(true) + .withCreateNotify(true) + .build(); + + Watch.Watcher watcher = watch.watch(key, option, new Watch.Listener() { + @Override + public void onNext(WatchResponse response) { + if (response.isCreatedNotify()) { + created.countDown(); + } + for (WatchEvent event : response.getEvents()) { + KeyValue item = event.getKeyValue(); + long revision = item.getModRevision(); + if (revision > endRevision) { + continue; + } + Map row = new LinkedHashMap<>(); + row.put("eventType", event.getEventType() == WatchEvent.EventType.DELETE ? "delete" : "put"); + row.put("revision", longString(revision)); + row.put( + "value", + event.getEventType() == WatchEvent.EventType.DELETE + ? null + : valueObject(item.getValue().getBytes()) + ); + KeyValue previous = event.getPrevKV(); + row.put( + "previousValue", + previous != null && previous.getVersion() > 0 + ? valueObject(previous.getValue().getBytes()) + : null + ); + row.put( + "metadata", + event.getEventType() == WatchEvent.EventType.DELETE && previous != null + ? metadata(previous) + : metadata(item) + ); + appendBoundedHistory(events, row, limit, truncated); + if (revision >= targetKeyRevision) { + completed.countDown(); + } + } + if (response.isProgressNotify() && response.getHeader().getRevision() >= endRevision) { + completed.countDown(); + } + } + + @Override + public void onError(Throwable throwable) { + failure.set(throwable); + completed.countDown(); + } + + @Override + public void onCompleted() { + completed.countDown(); + } + }); + + try { + if (!created.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("ETCD_HISTORY_TIMEOUT: watcher was not created"); + } + // For an existing exact key, its latest mod revision is an explicit + // replay boundary. This avoids relying on progress notifications, + // which older etcd/jetcd combinations do not consistently emit. + watcher.requestProgress(); + if (!completed.await(15, TimeUnit.SECONDS)) { + throw new IllegalStateException("ETCD_HISTORY_TIMEOUT: history replay did not reach the requested revision"); + } + } finally { + watcher.close(); + } + if (failure.get() != null) { + Throwable cause = rootCause(failure.get()); + if (cause instanceof io.etcd.jetcd.common.exception.CompactedException compacted) { + throw new IllegalStateException( + "ETCD_COMPACTED: requested history was compacted at revision " + compacted.getCompactedRevision() + ); + } + throw new IllegalStateException("ETCD_HISTORY_FAILED: " + safeMessage(cause)); + } + + List> page; + synchronized (events) { + page = new ArrayList<>(events); + } + page.sort(Comparator.comparingLong(row -> -Long.parseLong((String) row.get("revision")))); + Map result = new LinkedHashMap<>(); + result.put("events", page); + result.put("observedRevision", longString(endRevision)); + result.put("truncated", truncated.get()); + return result; + } + + static long historyStartRevision(Long requestedStart, long targetRevision) { + if (requestedStart != null) { + return Math.max(1L, requestedStart); + } + return Math.max(1L, targetRevision - HISTORY_DEFAULT_REVISION_WINDOW + 1L); + } + + static void appendBoundedHistory( + Deque events, + T event, + int limit, + AtomicBoolean truncated + ) { + synchronized (events) { + if (events.size() == limit) { + events.removeFirst(); + truncated.set(true); + } + events.addLast(event); + } + } + + private static Object status() throws Exception { + requireKv(); + Maintenance maintenance = client.getMaintenanceClient(); + Cluster cluster = client.getClusterClient(); + Map membersById = new HashMap<>(); + List endpoints = new ArrayList<>(connectedEndpoints); + try { + MemberListResponse memberList = cluster.listMember().get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + for (Member member : memberList.getMembers()) { + membersById.put(member.getId(), member); + for (java.net.URI uri : member.getClientURIs()) { + if (uri.getHost() == null || "0.0.0.0".equals(uri.getHost()) || "::".equals(uri.getHost())) { + continue; + } + String endpoint = uri.toString(); + if (!endpoints.contains(endpoint)) { + endpoints.add(endpoint); + } + } + } + } catch (Exception ignored) { + // Status still provides useful information for configured endpoints. + } + + List alarms = new ArrayList<>(); + try { + AlarmResponse alarmResponse = maintenance.listAlarms().get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + for (AlarmMember alarm : alarmResponse.getAlarms()) { + alarms.add(alarm.getAlarmType().name() + "@" + Long.toUnsignedString(alarm.getMemberId())); + } + } catch (Exception error) { + alarms.add("UNAVAILABLE: " + safeMessage(rootCause(error))); + } + + GetOption countOption = GetOption.newBuilder() + .withRange(ByteSequence.from(new byte[] {0})) + .withCountOnly(true) + .build(); + GetResponse countResponse = kv.get(ByteSequence.from(new byte[] {0}), countOption) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + + List> statusMembers = new ArrayList<>(); + Set observedMemberIds = new HashSet<>(); + String clusterId = null; + String revision = longString(countResponse.getHeader().getRevision()); + String leaderId = null; + Map> statusRequests = new LinkedHashMap<>(); + Map statusStartedNanos = new LinkedHashMap<>(); + for (String endpoint : endpoints) { + statusStartedNanos.put(endpoint, System.nanoTime()); + statusRequests.put(endpoint, maintenance.statusMember(endpoint)); + } + for (String endpoint : endpoints) { + Map row = new LinkedHashMap<>(); + row.put("endpoint", endpoint); + long started = statusStartedNanos.get(endpoint); + try { + StatusResponse memberStatus = statusRequests.get(endpoint) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + long latencyMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started); + long memberId = memberStatus.getHeader().getMemberId(); + Member member = membersById.get(memberId); + clusterId = unsignedLongString(memberStatus.getHeader().getClusterId()); + leaderId = unsignedLongString(memberStatus.getLeader()); + row.put("memberId", unsignedLongString(memberId)); + row.put("name", member == null ? null : member.getName()); + row.put("version", memberStatus.getVersion()); + row.put("leaderId", unsignedLongString(memberStatus.getLeader())); + row.put("revision", longString(memberStatus.getHeader().getRevision())); + row.put("raftTerm", longString(memberStatus.getRaftTerm())); + row.put("raftIndex", longString(memberStatus.getRaftIndex())); + row.put("raftAppliedIndex", longString(memberStatus.getRaftAppliedIndex())); + row.put("dbSize", longString(memberStatus.getDbSize())); + row.put("dbSizeInUse", longString(memberStatus.getDbSizeInUse())); + row.put("learner", memberStatus.isLearner()); + row.put("reachable", true); + row.put("latencyMs", latencyMs); + row.put("errors", memberStatus.getErrorList()); + if (!observedMemberIds.add(unsignedLongString(memberId))) { + continue; + } + } catch (Exception error) { + statusRequests.get(endpoint).cancel(true); + row.put("memberId", null); + row.put("name", null); + row.put("version", null); + row.put("leaderId", null); + row.put("revision", null); + row.put("raftTerm", null); + row.put("raftIndex", null); + row.put("raftAppliedIndex", null); + row.put("dbSize", null); + row.put("dbSizeInUse", null); + row.put("learner", false); + row.put("reachable", false); + row.put("latencyMs", null); + row.put("errors", List.of(safeMessage(rootCause(error)))); + } + statusMembers.add(row); + } + + Map result = new LinkedHashMap<>(); + result.put("clusterId", clusterId); + result.put("revision", revision); + result.put("leaderId", leaderId); + result.put("keyCount", longString(countResponse.getCount())); + result.put("alarms", alarms); + result.put("members", statusMembers); return result; } @@ -300,6 +669,9 @@ public final class EtcdAgent { case AgentProtocol.KV_METHOD_GET -> get(params); case AgentProtocol.KV_METHOD_PUT -> put(params); case AgentProtocol.KV_METHOD_DELETE -> delete(params); + case AgentProtocol.KV_METHOD_RENAME -> rename(params); + case AgentProtocol.KV_METHOD_HISTORY -> history(params); + case AgentProtocol.KV_METHOD_STATUS -> status(); case AgentProtocol.METHOD_DISCONNECT -> { closeClient(); yield Collections.singletonMap("ok", true); @@ -351,9 +723,7 @@ public final class EtcdAgent { } private static Client requireClient() { - if (client == null) { - throw new IllegalStateException("Not connected"); - } + if (client == null) throw new IllegalStateException("Not connected"); return client; } @@ -366,6 +736,7 @@ public final class EtcdAgent { client.close(); client = null; } + connectedEndpoints = Collections.emptyList(); } private static ByteSequence byteSequence(String value) { @@ -403,14 +774,44 @@ public final class EtcdAgent { private static Map metadata(KeyValue item) { Map metadata = new LinkedHashMap<>(); - metadata.put("createRevision", item.getCreateRevision()); - metadata.put("modRevision", item.getModRevision()); - metadata.put("version", item.getVersion()); - metadata.put("lease", item.getLease()); + metadata.put("createRevision", longString(item.getCreateRevision())); + metadata.put("modRevision", longString(item.getModRevision())); + metadata.put("version", longString(item.getVersion())); + metadata.put("lease", longString(item.getLease())); metadata.put("valueSize", item.getValue().size()); return metadata; } + private static Map metadataWithTtl(KeyValue item) throws Exception { + Map metadata = metadata(item); + if (item.getLease() > 0) { + LeaseTimeToLiveResponse lease = requireClient().getLeaseClient() + .timeToLive(item.getLease(), LeaseOption.DEFAULT) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + metadata.put("ttl", lease.getTTL()); + } + return metadata; + } + + private static ByteSequence keyBytes(JsonObject params) { + JsonElement encoded = params.get("keyBytes"); + if (encoded != null && encoded.isJsonObject()) { + return ByteSequence.from(parseValue(encoded.getAsJsonObject())); + } + JsonElement key = params.get("key"); + if (key == null || key.isJsonNull()) { + throw new IllegalArgumentException("Key is required"); + } + return byteSequence(key.getAsString()); + } + + private static Map bytesObject(byte[] bytes) { + Map result = new LinkedHashMap<>(); + result.put("encoding", "base64"); + result.put("data", Base64.getEncoder().encodeToString(bytes)); + return result; + } + private static Map valueObject(byte[] bytes) { Map value = new LinkedHashMap<>(); String utf8 = strictUtf8(bytes); @@ -476,6 +877,32 @@ public final class EtcdAgent { return element == null || element.isJsonNull() ? fallback : element.getAsBoolean(); } + private static Long longOrNull(JsonObject object, String key) { + JsonElement element = object.get(key); + return element == null || element.isJsonNull() ? null : element.getAsLong(); + } + + private static String longString(long value) { + return Long.toString(value); + } + + private static String unsignedLongString(long value) { + return Long.toUnsignedString(value); + } + + private static Throwable rootCause(Throwable throwable) { + Throwable current = throwable; + while (current.getCause() != null && current.getCause() != current) { + current = current.getCause(); + } + return current; + } + + private static String safeMessage(Throwable throwable) { + String message = throwable == null ? null : throwable.getMessage(); + return message == null || message.isBlank() ? String.valueOf(throwable) : message; + } + private static String firstNonBlank(String... values) { for (String value : values) { if (value != null && !value.isBlank()) { diff --git a/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java b/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java index b6de4f8cd7..6b41fd846e 100644 --- a/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java +++ b/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java @@ -5,6 +5,7 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import io.etcd.jetcd.ByteSequence; import io.etcd.jetcd.KV; +import io.etcd.jetcd.Lease; import io.etcd.jetcd.Txn; import io.etcd.jetcd.api.KeyValue; import io.etcd.jetcd.api.RangeResponse; @@ -18,6 +19,8 @@ import java.util.Arrays; import java.util.Deque; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -34,6 +37,10 @@ final class EtcdAgentTest { Assertions.assertEquals(1, result.get("protocolVersion").getAsInt()); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_ttl\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_cas\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_list_values\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_status\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_history\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"connect\""))); } @@ -70,6 +77,38 @@ final class EtcdAgentTest { Assertions.assertEquals("Not connected", error.get("message").getAsString()); } + @Test + void historyMethodIsRegistered() { + String response = EtcdAgent.handleRequest( + "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"kv_history\",\"params\":{\"key\":\"/app/name\",\"limit\":100}}" + ); + + JsonObject error = JsonParser.parseString(response).getAsJsonObject().getAsJsonObject("error"); + + Assertions.assertEquals(-1, error.get("code").getAsInt()); + Assertions.assertEquals("Not connected", error.get("message").getAsString()); + } + + @Test + void historyDefaultsToABoundedRevisionWindow() { + Assertions.assertEquals(1L, EtcdAgent.historyStartRevision(null, 100L)); + Assertions.assertEquals(10_001L, EtcdAgent.historyStartRevision(null, 20_000L)); + Assertions.assertEquals(42L, EtcdAgent.historyStartRevision(42L, 20_000L)); + } + + @Test + void historyRetainsOnlyTheNewestRequestedEvents() { + Deque events = new ArrayDeque<>(); + AtomicBoolean truncated = new AtomicBoolean(); + + for (int value = 1; value <= 5; value++) { + EtcdAgent.appendBoundedHistory(events, value, 3, truncated); + } + + Assertions.assertEquals(List.of(3, 4, 5), List.copyOf(events)); + Assertions.assertTrue(truncated.get()); + } + @Test void preserveLeaseRetriesWithTheLatestLeaseAfterConcurrentChange() throws Exception { AtomicInteger getCalls = new AtomicInteger(); @@ -141,6 +180,36 @@ final class EtcdAgentTest { Assertions.assertEquals(3, txnCalls.get()); } + @Test + void grantedLeaseIsRevokedWhenWriteFailsAfterCreation() { + AtomicInteger revokeCalls = new AtomicInteger(); + AtomicInteger revokedLeaseId = new AtomicInteger(); + Lease lease = (Lease) Proxy.newProxyInstance( + Lease.class.getClassLoader(), + new Class[] { Lease.class }, + (proxy, method, args) -> { + if (method.getName().equals("revoke")) { + revokeCalls.incrementAndGet(); + revokedLeaseId.set(Math.toIntExact((long) args[0])); + return CompletableFuture.completedFuture(null); + } + if (method.getName().equals("close")) return null; + throw new UnsupportedOperationException(method.getName()); + } + ); + + TimeoutException error = Assertions.assertThrows( + TimeoutException.class, + () -> EtcdAgent.cleanUpGrantedLeaseOnFailure(123, lease, () -> { + throw new TimeoutException("write timed out"); + }) + ); + + Assertions.assertEquals("write timed out", error.getMessage()); + Assertions.assertEquals(1, revokeCalls.get()); + Assertions.assertEquals(123, revokedLeaseId.get()); + } + private static KV scriptedKv( AtomicInteger getCalls, AtomicInteger txnCalls, diff --git a/apps/desktop/src/components/etcd/EtcdDashboard.vue b/apps/desktop/src/components/etcd/EtcdDashboard.vue new file mode 100644 index 0000000000..9a6856e2b8 --- /dev/null +++ b/apps/desktop/src/components/etcd/EtcdDashboard.vue @@ -0,0 +1,584 @@ + + + diff --git a/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue b/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue index da92670d89..61fed498c3 100644 --- a/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue +++ b/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue @@ -1,12 +1,63 @@ diff --git a/apps/desktop/src/components/etcd/__tests__/EtcdDashboardConcurrency.spec.ts b/apps/desktop/src/components/etcd/__tests__/EtcdDashboardConcurrency.spec.ts new file mode 100644 index 0000000000..147e0c598d --- /dev/null +++ b/apps/desktop/src/components/etcd/__tests__/EtcdDashboardConcurrency.spec.ts @@ -0,0 +1,15 @@ +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; + +const dashboardSource = readFileSync(new URL("../EtcdDashboard.vue", import.meta.url), "utf8"); + +describe("EtcdDashboard refresh concurrency", () => { + it("skips overlapping refreshes for the same connection", () => { + expect(dashboardSource).toContain("if (loading.value && loadingConnectionId === connectionId) return"); + }); + + it("discards responses from an obsolete load generation", () => { + expect(dashboardSource).toContain("const generation = ++loadGeneration"); + expect(dashboardSource).toContain("if (generation !== loadGeneration || props.connectionId !== connectionId) return"); + }); +}); diff --git a/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts b/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts index fb2da1aa56..f4bc7b3b3f 100644 --- a/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts +++ b/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts @@ -5,6 +5,15 @@ import { createApp, defineComponent, h, nextTick, type App } from "vue"; const backend = vi.hoisted(() => ({ etcdSupportsTtl: vi.fn(), + etcdListPrefix: vi.fn(), + etcdGet: vi.fn(), + etcdPut: vi.fn(), + etcdDelete: vi.fn(), + etcdRename: vi.fn(), + etcdHistory: vi.fn(), + etcdGetCalls: [] as unknown[][], + capturedKvBrowserApi: null as any, + selectKeyCalls: [] as unknown[], })); vi.mock("vue-i18n", () => ({ @@ -12,24 +21,41 @@ vi.mock("vue-i18n", () => ({ })); vi.mock("@/lib/backend/api", () => ({ - etcdListPrefix: vi.fn(), - etcdGet: vi.fn(), - etcdPut: vi.fn(), - etcdDelete: vi.fn(), + etcdListPrefix: backend.etcdListPrefix, + etcdGet: backend.etcdGet, + etcdPut: backend.etcdPut, + etcdDelete: backend.etcdDelete, + etcdRename: backend.etcdRename, + etcdHistory: backend.etcdHistory, etcdSupportsTtl: backend.etcdSupportsTtl, })); +vi.mock("@/stores/connectionStore", () => ({ + useConnectionStore: () => ({ + connections: [], + getConfig: () => undefined, + }), +})); + vi.mock("@/components/kv/KvKeyBrowser.vue", () => ({ default: defineComponent({ name: "KvKeyBrowserStub", + inheritAttrs: false, props: { + api: { type: Object, required: true }, + connectionId: { type: String, required: true }, supportsTtl: { type: Boolean, default: false }, ttlCapabilityKnown: { type: Boolean, default: true }, }, emits: ["refreshRequested"], setup(props, { emit, expose }) { + backend.capturedKvBrowserApi = props.api; expose({ focusSearch: () => true, + selectKey: (route: unknown) => { + backend.selectKeyCalls.push(route); + return Promise.resolve(); + }, refresh: () => { emit("refreshRequested"); return true; @@ -73,6 +99,16 @@ async function mountBrowser() { beforeEach(() => { vi.useFakeTimers(); backend.etcdSupportsTtl.mockReset(); + backend.etcdListPrefix.mockReset(); + backend.etcdGet.mockReset(); + backend.etcdPut.mockReset(); + backend.etcdDelete.mockReset(); + backend.etcdRename.mockReset(); + backend.etcdHistory.mockReset(); + backend.etcdListPrefix.mockResolvedValue({ keys: [], continuation: null, revision: "1" }); + backend.etcdGetCalls.length = 0; + backend.capturedKvBrowserApi = null; + backend.selectKeyCalls.length = 0; }); afterEach(() => { @@ -129,3 +165,115 @@ describe("EtcdKeyBrowser TTL capability recovery", () => { expect(browser.dataset.capabilityKnown).toBe("true"); }); }); + +describe("EtcdKeyBrowser byte-identity routing", () => { + it("keeps colliding display keys on distinct keyBytes routes", async () => { + backend.etcdListPrefix.mockResolvedValue({ + keys: [ + { + key: "/w==", + keyBytes: { encoding: "base64", data: "/w==" }, + value: { encoding: "utf8", data: "binary" }, + modRevision: "1", + }, + { + key: "[base64:/w==]", + keyBytes: { encoding: "utf8", data: "[base64:/w==]" }, + value: { encoding: "utf8", data: "utf8" }, + modRevision: "2", + }, + ], + continuation: null, + revision: "7", + }); + backend.etcdGet.mockImplementation(async (_connectionId, key, options) => ({ + found: true, + key, + keyBytes: options?.keyBytes ?? null, + value: { encoding: "utf8", data: "value" }, + metadata: { modRevision: "9" }, + })); + backend.etcdPut.mockResolvedValue({ revision: 10, key: "[base64:/w==]" }); + backend.etcdDelete.mockResolvedValue({ deleted: 1 }); + backend.etcdRename.mockResolvedValue({ renamed: true, revision: "11" }); + backend.etcdHistory.mockResolvedValue({ events: [] }); + + await mountBrowser(); + expect(backend.capturedKvBrowserApi).toBeTruthy(); + + const { keys } = await backend.capturedKvBrowserApi.listPrefix("etcd-1", "", 20); + expect(keys).toHaveLength(2); + const [binaryKey, utf8Key] = keys; + expect(binaryKey.key).toBe("[base64:/w==]"); + expect(utf8Key.key).toBe("[base64:/w==]"); + expect(binaryKey.keyIdentity).not.toBe(utf8Key.keyIdentity); + + await backend.capturedKvBrowserApi.get("etcd-1", binaryKey.key, { keyBytes: binaryKey.keyBytes }); + await backend.capturedKvBrowserApi.put("etcd-1", binaryKey.key, { encoding: "utf8", data: "x" }, { keyBytes: binaryKey.keyBytes }); + await backend.capturedKvBrowserApi.deleteKey("etcd-1", utf8Key.key, { keyBytes: utf8Key.keyBytes }); + await backend.capturedKvBrowserApi.rename?.("etcd-1", { key: binaryKey.key, keyBytes: binaryKey.keyBytes, newKey: "renamed", expectedModRevision: "9" }); + await backend.capturedKvBrowserApi.history?.("etcd-1", { key: utf8Key.key, keyBytes: utf8Key.keyBytes, limit: 10 }); + await backend.capturedKvBrowserApi.exportScope?.("etcd-1", { path: binaryKey.key, kind: "key", keyBytes: binaryKey.keyBytes }); + + expect(backend.etcdGet).toHaveBeenCalledWith("etcd-1", "[base64:/w==]", expect.objectContaining({ keyBytes: binaryKey.keyBytes })); + expect(backend.etcdPut).toHaveBeenCalledWith("etcd-1", "[base64:/w==]", { encoding: "utf8", data: "x" }, expect.objectContaining({ keyBytes: binaryKey.keyBytes })); + expect(backend.etcdDelete).toHaveBeenCalledWith("etcd-1", "[base64:/w==]", expect.objectContaining({ keyBytes: utf8Key.keyBytes })); + expect(backend.etcdRename).toHaveBeenCalledWith("etcd-1", expect.objectContaining({ key: "[base64:/w==]", keyBytes: binaryKey.keyBytes })); + expect(backend.etcdHistory).toHaveBeenCalledWith("etcd-1", expect.objectContaining({ key: "[base64:/w==]", keyBytes: utf8Key.keyBytes })); + expect(backend.etcdGet.mock.calls.at(-1)?.[2]).toEqual(expect.objectContaining({ keyBytes: binaryKey.keyBytes })); + }); + + it("opens search results with the selected keyBytes", async () => { + backend.etcdListPrefix.mockResolvedValue({ + keys: [ + { + key: "/w==", + keyBytes: { encoding: "base64", data: "/w==" }, + value: { encoding: "utf8", data: "match" }, + modRevision: "1", + }, + { + key: "[base64:/w==]", + keyBytes: { encoding: "utf8", data: "[base64:/w==]" }, + value: { encoding: "utf8", data: "match" }, + modRevision: "2", + }, + ], + continuation: null, + revision: "7", + }); + + const root = document.createElement("div"); + document.body.appendChild(root); + const app = createApp(EtcdKeyBrowser, { connectionId: "etcd-1" }); + app.mount(root); + await flushUi(); + + const searchButton = [...root.querySelectorAll("button")].find((button) => button.textContent?.includes("etcd.globalSearch")) as HTMLButtonElement | undefined; + expect(searchButton).toBeTruthy(); + searchButton!.click(); + await flushUi(); + + const input = root.querySelector('input[placeholder="etcd.searchPlaceholder"]') as HTMLInputElement | null; + expect(input).toBeTruthy(); + input.value = "match"; + input.dispatchEvent(new Event("input")); + await flushUi(); + + const searchSubmit = [...root.querySelectorAll("button")].filter((button) => button.textContent?.includes("etcd.globalSearch")).at(-1) as HTMLButtonElement | undefined; + searchSubmit?.click(); + await flushUi(); + + const rows = [...root.querySelectorAll("tbody tr")]; + expect(rows).toHaveLength(2); + rows[0].dispatchEvent(new MouseEvent("dblclick", { bubbles: true })); + rows[1].dispatchEvent(new MouseEvent("dblclick", { bubbles: true })); + await flushUi(); + + expect(backend.selectKeyCalls).toHaveLength(2); + expect(backend.selectKeyCalls[0]).toEqual(expect.objectContaining({ keyBytes: { encoding: "base64", data: "/w==" } })); + expect(backend.selectKeyCalls[1]).toEqual(expect.objectContaining({ keyBytes: { encoding: "utf8", data: "[base64:/w==]" } })); + app.unmount(); + root.remove(); + }); +}); diff --git a/apps/desktop/src/components/kv/KvKeyBrowser.vue b/apps/desktop/src/components/kv/KvKeyBrowser.vue index 8158241a42..ba5c5ed1ec 100644 --- a/apps/desktop/src/components/kv/KvKeyBrowser.vue +++ b/apps/desktop/src/components/kv/KvKeyBrowser.vue @@ -1,8 +1,10 @@