diff --git a/README.zh-CN.md b/README.zh-CN.md index 4d26483bf8..d53a6cecb9 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -416,6 +416,7 @@ DBX 是免费开源项目，但持续维护、数据库兼容性测试、基础
可以离线使用吗？ + 可以。桌面端完全支持离线使用。内网环境安装驱动时，可在有网机器打开[离线驱动下载页](https://dbxio.com/cn/drivers)下载离线驱动包，传输到内网机器后，在 DBX 的「设置 > 驱动管理」中导入。AI 功能需要网络访问模型端点（或通过 Ollama 使用本地模型）。
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 3bba50c452..9b7ce35fe2 100644 --- a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java +++ b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java @@ -76,6 +76,7 @@ public final class AgentProtocol { public static final String CAPABILITY_TRANSACTION = "transaction"; 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_MULTI_SESSION = "multi_session"; public static final List CAPABILITIES = Collections.unmodifiableList(Arrays.asList( @@ -96,7 +97,8 @@ public final class AgentProtocol { CAPABILITY_PAGED_QUERY, CAPABILITY_TRANSACTION, CAPABILITY_DDL, - CAPABILITY_KV + CAPABILITY_KV, + CAPABILITY_KV_TTL )); public static final List COMMON_METHODS = Collections.unmodifiableList(Arrays.asList( diff --git a/agents/common/src/main/resources/agent-protocol-v1.json b/agents/common/src/main/resources/agent-protocol-v1.json index cec73f1784..63cd20bf50 100644 --- a/agents/common/src/main/resources/agent-protocol-v1.json +++ b/agents/common/src/main/resources/agent-protocol-v1.json @@ -14,7 +14,8 @@ "paged_query", "transaction", "ddl", - "kv" + "kv", + "kv_ttl" ], "capabilities": [ "connect", diff --git a/agents/common/src/main/resources/agent-protocol-v2.json b/agents/common/src/main/resources/agent-protocol-v2.json index 96798e8e6a..a00b78a1c6 100644 --- a/agents/common/src/main/resources/agent-protocol-v2.json +++ b/agents/common/src/main/resources/agent-protocol-v2.json @@ -15,6 +15,7 @@ "transaction", "ddl", "kv", + "kv_ttl", "multi_session" ], "capabilities": [ 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 c29286f047..9c33cede80 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 @@ -14,7 +14,14 @@ 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.kv.TxnResponse; +import io.etcd.jetcd.lease.LeaseGrantResponse; +import io.etcd.jetcd.lease.LeaseTimeToLiveResponse; +import io.etcd.jetcd.op.Cmp; +import io.etcd.jetcd.op.CmpTarget; +import io.etcd.jetcd.op.Op; import io.etcd.jetcd.options.GetOption; +import io.etcd.jetcd.options.LeaseOption; import io.etcd.jetcd.options.PutOption; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; @@ -38,10 +45,12 @@ 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 List CAPABILITIES = Collections.unmodifiableList(Arrays.asList( AgentProtocol.CAPABILITY_CONNECT, AgentProtocol.CAPABILITY_TEST_CONNECTION, - AgentProtocol.CAPABILITY_KV + AgentProtocol.CAPABILITY_KV, + AgentProtocol.CAPABILITY_KV_TTL )); private static Client client; private static KV kv; @@ -162,7 +171,11 @@ public final class EtcdAgent { private static Object get(JsonObject params) throws Exception { KV active = requireKv(); String key = params.get("key").getAsString(); - GetResponse response = active.get(byteSequence(key)).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + 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); Map result = new LinkedHashMap<>(); if (response.getKvs().isEmpty()) { result.put("found", false); @@ -172,26 +185,101 @@ 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", valueObject(item.getValue().getBytes())); - result.put("metadata", metadata(item)); + result.put("value", metadataOnly ? null : valueObject(item.getValue().getBytes())); + result.put("metadata", metadata); return result; } private static Object put(JsonObject params) throws Exception { KV active = requireKv(); + Client activeClient = requireClient(); String key = params.get("key").getAsString(); byte[] value = parseValue(params.getAsJsonObject("value")); + ByteSequence keyBytes = byteSequence(key); + ByteSequence valueBytes = ByteSequence.from(value); JsonElement leaseElement = params.get("lease"); - PutResponse response; - if (leaseElement != null && !leaseElement.isJsonNull()) { - PutOption option = PutOption.newBuilder().withLeaseId(leaseElement.getAsLong()).build(); - response = active.put(byteSequence(key), ByteSequence.from(value), option).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); - } else { - response = active.put(byteSequence(key), ByteSequence.from(value)).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + JsonElement ttlElement = params.get("ttl"); + boolean hasLease = leaseElement != null && !leaseElement.isJsonNull(); + boolean hasTtl = ttlElement != null && !ttlElement.isJsonNull(); + boolean preserveLease = boolOrDefault(params, "preserveLease", false); + if ((hasLease && hasTtl) || (preserveLease && (hasLease || hasTtl))) { + throw new IllegalArgumentException("lease, ttl, and preserveLease cannot be specified together"); } - return Collections.singletonMap("revision", response.getHeader().getRevision()); + + if (preserveLease) { + long revision = putPreservingLease(active, keyBytes, valueBytes); + return Collections.singletonMap("revision", revision); + } + + Long leaseId = null; + 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); + grantedLeaseId = grant.getID(); + leaseId = grantedLeaseId; + } else if (hasLease) { + leaseId = leaseElement.getAsLong(); + } + + 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) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + return Collections.singletonMap("revision", response.getHeader().getRevision()); + } catch (Exception error) { + if (grantedLeaseId != 0) { + try { + activeClient.getLeaseClient().revoke(grantedLeaseId) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (Exception revokeError) { + error.addSuppressed(revokeError); + } + } + throw error; + } + } + + 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); + if (existing.getKvs().isEmpty() || existing.getKvs().get(0).getLease() <= 0) { + throw new IllegalStateException("Cannot preserve lease: key does not exist or has no lease"); + } + + KeyValue current = existing.getKvs().get(0); + PutOption option = PutOption.builder().withLeaseId(current.getLease()).build(); + TxnResponse response = active.txn() + .If(new Cmp(key, Cmp.Op.EQUAL, CmpTarget.modRevision(current.getModRevision()))) + .Then(Op.put(key, value, option)) + .commit() + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (response.isSucceeded()) { + return response.getHeader().getRevision(); + } + } + + throw new IllegalStateException("Cannot preserve lease: key changed concurrently; retry the save"); } private static Object delete(JsonObject params) throws Exception { @@ -262,6 +350,13 @@ public final class EtcdAgent { return kv; } + private static Client requireClient() { + if (client == null) { + throw new IllegalStateException("Not connected"); + } + return client; + } + private static void closeClient() { if (kv != null) { kv.close(); 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 65c205df8a..b6de4f8cd7 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 @@ -1,8 +1,24 @@ package com.dbx.agent.etcd; +import com.google.protobuf.ByteString; import com.google.gson.JsonObject; import com.google.gson.JsonParser; +import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.KV; +import io.etcd.jetcd.Txn; +import io.etcd.jetcd.api.KeyValue; +import io.etcd.jetcd.api.RangeResponse; +import io.etcd.jetcd.api.ResponseHeader; +import io.etcd.jetcd.kv.GetResponse; +import io.etcd.jetcd.kv.TxnResponse; +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Deque; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -17,6 +33,7 @@ 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("\"connect\""))); } @@ -52,4 +69,155 @@ final class EtcdAgentTest { Assertions.assertEquals(-1, error.get("code").getAsInt()); Assertions.assertEquals("Not connected", error.get("message").getAsString()); } + + @Test + void preserveLeaseRetriesWithTheLatestLeaseAfterConcurrentChange() throws Exception { + AtomicInteger getCalls = new AtomicInteger(); + AtomicInteger txnCalls = new AtomicInteger(); + KV active = scriptedKv( + getCalls, + txnCalls, + List.of(getResponse("/app/name", 10, 101), getResponse("/app/name", 11, 202)), + List.of(txnResponse(false, 10), txnResponse(true, 12)) + ); + + long revision = EtcdAgent.putPreservingLease(active, bytes("/app/name"), bytes("updated")); + + Assertions.assertEquals(12, revision); + Assertions.assertEquals(2, getCalls.get()); + Assertions.assertEquals(2, txnCalls.get()); + } + + @Test + void preserveLeaseRejectsMissingOrUnleasedKeys() { + KV missing = scriptedKv( + new AtomicInteger(), + new AtomicInteger(), + List.of(emptyGetResponse()), + List.of() + ); + KV unleased = scriptedKv( + new AtomicInteger(), + new AtomicInteger(), + List.of(getResponse("/app/name", 10, 0)), + List.of() + ); + + IllegalStateException missingError = Assertions.assertThrows( + IllegalStateException.class, + () -> EtcdAgent.putPreservingLease(missing, bytes("/app/name"), bytes("updated")) + ); + IllegalStateException unleasedError = Assertions.assertThrows( + IllegalStateException.class, + () -> EtcdAgent.putPreservingLease(unleased, bytes("/app/name"), bytes("updated")) + ); + + Assertions.assertEquals("Cannot preserve lease: key does not exist or has no lease", missingError.getMessage()); + Assertions.assertEquals("Cannot preserve lease: key does not exist or has no lease", unleasedError.getMessage()); + } + + @Test + void preserveLeaseFailsClearlyAfterBoundedConcurrentChanges() { + AtomicInteger getCalls = new AtomicInteger(); + AtomicInteger txnCalls = new AtomicInteger(); + KV active = scriptedKv( + getCalls, + txnCalls, + List.of( + getResponse("/app/name", 10, 101), + getResponse("/app/name", 11, 202), + getResponse("/app/name", 12, 303) + ), + List.of(txnResponse(false, 10), txnResponse(false, 11), txnResponse(false, 12)) + ); + + IllegalStateException error = Assertions.assertThrows( + IllegalStateException.class, + () -> EtcdAgent.putPreservingLease(active, bytes("/app/name"), bytes("updated")) + ); + + Assertions.assertEquals("Cannot preserve lease: key changed concurrently; retry the save", error.getMessage()); + Assertions.assertEquals(3, getCalls.get()); + Assertions.assertEquals(3, txnCalls.get()); + } + + private static KV scriptedKv( + AtomicInteger getCalls, + AtomicInteger txnCalls, + List getResponses, + List txnResponses + ) { + Deque gets = new ArrayDeque<>(getResponses); + Deque txns = new ArrayDeque<>(txnResponses); + return (KV) Proxy.newProxyInstance( + KV.class.getClassLoader(), + new Class[] { KV.class }, + (proxy, method, args) -> { + if (method.getName().equals("get")) { + getCalls.incrementAndGet(); + return CompletableFuture.completedFuture(gets.removeFirst()); + } + if (method.getName().equals("txn")) { + return scriptedTxn(txnCalls, txns); + } + if (method.getName().equals("close")) { + return null; + } + throw new UnsupportedOperationException(method.getName()); + } + ); + } + + private static Txn scriptedTxn(AtomicInteger txnCalls, Deque responses) { + return (Txn) Proxy.newProxyInstance( + Txn.class.getClassLoader(), + new Class[] { Txn.class }, + (proxy, method, args) -> { + if (Arrays.asList("If", "Then", "Else").contains(method.getName())) { + return proxy; + } + if (method.getName().equals("commit")) { + txnCalls.incrementAndGet(); + return CompletableFuture.completedFuture(responses.removeFirst()); + } + throw new UnsupportedOperationException(method.getName()); + } + ); + } + + private static ByteSequence bytes(String value) { + return ByteSequence.from(value, StandardCharsets.UTF_8); + } + + private static GetResponse getResponse(String key, long modRevision, long lease) { + KeyValue keyValue = KeyValue.newBuilder() + .setKey(ByteString.copyFromUtf8(key)) + .setModRevision(modRevision) + .setLease(lease) + .build(); + RangeResponse response = RangeResponse.newBuilder() + .setHeader(header(modRevision)) + .addKvs(keyValue) + .build(); + return new GetResponse(response, ByteSequence.EMPTY); + } + + private static GetResponse emptyGetResponse() { + return new GetResponse( + RangeResponse.newBuilder().setHeader(header(1)).build(), + ByteSequence.EMPTY + ); + } + + private static TxnResponse txnResponse(boolean succeeded, long revision) { + io.etcd.jetcd.api.TxnResponse response = io.etcd.jetcd.api.TxnResponse.newBuilder() + .setHeader(header(revision)) + .setSucceeded(succeeded) + .build(); + return new TxnResponse(response, ByteSequence.EMPTY); + } + + private static ResponseHeader header(long revision) { + return ResponseHeader.newBuilder().setRevision(revision).build(); + } } diff --git a/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue b/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue index 29249dea52..da92670d89 100644 --- a/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue +++ b/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue @@ -1,5 +1,5 @@ diff --git a/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts b/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts new file mode 100644 index 0000000000..fb2da1aa56 --- /dev/null +++ b/apps/desktop/src/components/etcd/__tests__/EtcdKeyBrowser.spec.ts @@ -0,0 +1,131 @@ +// @vitest-environment happy-dom + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createApp, defineComponent, h, nextTick, type App } from "vue"; + +const backend = vi.hoisted(() => ({ + etcdSupportsTtl: vi.fn(), +})); + +vi.mock("vue-i18n", () => ({ + useI18n: () => ({ t: (key: string) => key }), +})); + +vi.mock("@/lib/backend/api", () => ({ + etcdListPrefix: vi.fn(), + etcdGet: vi.fn(), + etcdPut: vi.fn(), + etcdDelete: vi.fn(), + etcdSupportsTtl: backend.etcdSupportsTtl, +})); + +vi.mock("@/components/kv/KvKeyBrowser.vue", () => ({ + default: defineComponent({ + name: "KvKeyBrowserStub", + props: { + supportsTtl: { type: Boolean, default: false }, + ttlCapabilityKnown: { type: Boolean, default: true }, + }, + emits: ["refreshRequested"], + setup(props, { emit, expose }) { + expose({ + focusSearch: () => true, + refresh: () => { + emit("refreshRequested"); + return true; + }, + }); + return () => + h( + "button", + { + id: "kv-browser-stub", + "data-supports-ttl": String(props.supportsTtl), + "data-capability-known": String(props.ttlCapabilityKnown), + onClick: () => emit("refreshRequested"), + }, + "refresh", + ); + }, + }), +})); + +import EtcdKeyBrowser from "@/components/etcd/EtcdKeyBrowser.vue"; + +let app: App | null = null; +let root: HTMLDivElement | null = null; + +async function flushUi() { + await Promise.resolve(); + await Promise.resolve(); + await nextTick(); +} + +async function mountBrowser() { + root = document.createElement("div"); + document.body.appendChild(root); + app = createApp(EtcdKeyBrowser, { connectionId: "etcd-1" }); + app.mount(root); + await flushUi(); + return root.querySelector("#kv-browser-stub")!; +} + +beforeEach(() => { + vi.useFakeTimers(); + backend.etcdSupportsTtl.mockReset(); +}); + +afterEach(() => { + app?.unmount(); + app = null; + root?.remove(); + root = null; + vi.useRealTimers(); +}); + +describe("EtcdKeyBrowser TTL capability recovery", () => { + it("keeps a transient probe failure unknown and retries automatically", async () => { + backend.etcdSupportsTtl.mockRejectedValueOnce(new Error("Agent is reconnecting")).mockResolvedValue(true); + + const browser = await mountBrowser(); + expect(browser.dataset.supportsTtl).toBe("false"); + expect(browser.dataset.capabilityKnown).toBe("false"); + + await vi.advanceTimersByTimeAsync(5000); + await flushUi(); + + expect(backend.etcdSupportsTtl).toHaveBeenCalledTimes(2); + expect(browser.dataset.supportsTtl).toBe("true"); + expect(browser.dataset.capabilityKnown).toBe("true"); + }); + + it("rechecks a confirmed unsupported Agent when the user refreshes", async () => { + backend.etcdSupportsTtl.mockResolvedValueOnce(false).mockResolvedValueOnce(true); + + const browser = await mountBrowser(); + expect(browser.dataset.supportsTtl).toBe("false"); + expect(browser.dataset.capabilityKnown).toBe("true"); + + browser.click(); + await flushUi(); + + expect(backend.etcdSupportsTtl).toHaveBeenCalledTimes(2); + expect(browser.dataset.supportsTtl).toBe("true"); + expect(browser.dataset.capabilityKnown).toBe("true"); + }); + + it("keeps the last confirmed capability through a transient later failure", async () => { + backend.etcdSupportsTtl.mockResolvedValueOnce(true).mockRejectedValueOnce(new Error("temporary failure")); + + const browser = await mountBrowser(); + expect(browser.dataset.supportsTtl).toBe("true"); + expect(browser.dataset.capabilityKnown).toBe("true"); + + await vi.advanceTimersByTimeAsync(5000); + await flushUi(); + + expect(backend.etcdSupportsTtl).toHaveBeenCalledTimes(2); + expect(browser.dataset.supportsTtl).toBe("true"); + expect(browser.dataset.capabilityKnown).toBe("true"); + }); +}); diff --git a/apps/desktop/src/components/kv/KvKeyBrowser.vue b/apps/desktop/src/components/kv/KvKeyBrowser.vue index cbef150748..8158241a42 100644 --- a/apps/desktop/src/components/kv/KvKeyBrowser.vue +++ b/apps/desktop/src/components/kv/KvKeyBrowser.vue @@ -1,5 +1,5 @@ @@ -598,7 +846,7 @@ defineExpose({ focusSearch, refresh }); -