feat(moonzero): etcd v3 KV protobuf messages — the base of a real etcd client.

The registry models etcd v3 semantics but talks to no live etcd; the first step to a real network client is the etcdserverpb KV wire messages. KeyValue / RangeRequest / RangeResponse / PutRequest now encode and decode over moonrpc's protobuf runtime with the field numbers etcd's rpc.proto fixes, so they are byte-compatible with a real etcd server. Verified against reference vectors CPython's protobuf produced for the same messages (mutation-checked on a field number) and round-tripped on all four backends; the dependency on moonrpc is bumped to 0.7.0 for its PbWriter/PbReader.

Signed-off-by: 林晨 (Leo Cheng) <chengkelfan@qq.com>

Author: Lfan-ke

etcd.mbt

@@ -0,0 +1,217 @@
+// The etcd v3 KV RPC message layer (`etcdserverpb`), the first foundation of a real
+// etcd network client for service discovery (← go-zero's `discov`, which talks to a
+// live etcd over gRPC). The messages are self-built on moonrpc's protobuf runtime —
+// the same `PbWriter` / `PbReader` the gRPC transport uses — with the field numbers
+// fixed by etcd's `rpc.proto`, so a `KeyValue` / `RangeRequest` / `PutRequest` encodes
+// byte-for-byte to what a real etcd server expects. The gRPC calls over an
+// `@moonrpc.Channel` build on top of this.
+
+///|
+/// An etcd `KeyValue` (`mvccpb.KeyValue`): a key, its value, the create/mod revisions
+/// and version that track its history, and the lease it is attached to.
+pub(all) struct EtcdKeyValue {
+  key : Bytes
+  create_revision : Int64
+  mod_revision : Int64
+  version : Int64
+  value : Bytes
+  lease : Int64
+} derive(Eq)
+
+///|
+/// The empty key/value with all-zero metadata.
+pub fn EtcdKeyValue::empty() -> EtcdKeyValue {
+  {
+    key: b"",
+    create_revision: 0,
+    mod_revision: 0,
+    version: 0,
+    value: b"",
+    lease: 0,
+  }
+}
+
+///|
+/// Encode a `KeyValue` to its protobuf wire bytes (field numbers per etcd
+/// `mvccpb.proto`: key=1, create_revision=2, mod_revision=3, version=4, value=5,
+/// lease=6). Proto3 default (empty / zero) fields are omitted.
+pub fn EtcdKeyValue::encode(self : EtcdKeyValue) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.key.length() > 0 {
+    w.bytes_(1, self.key)
+  }
+  if self.create_revision != 0 {
+    w.int64(2, self.create_revision)
+  }
+  if self.mod_revision != 0 {
+    w.int64(3, self.mod_revision)
+  }
+  if self.version != 0 {
+    w.int64(4, self.version)
+  }
+  if self.value.length() > 0 {
+    w.bytes_(5, self.value)
+  }
+  if self.lease != 0 {
+    w.int64(6, self.lease)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `KeyValue` from protobuf wire bytes; unknown fields are skipped.
+pub fn EtcdKeyValue::decode(
+  data : Bytes,
+) -> EtcdKeyValue raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut key = b""
+  let mut create_revision = 0L
+  let mut mod_revision = 0L
+  let mut version = 0L
+  let mut value = b""
+  let mut lease = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      1 => key = r.read_bytes()
+      2 => create_revision = r.read_int64()
+      3 => mod_revision = r.read_int64()
+      4 => version = r.read_int64()
+      5 => value = r.read_bytes()
+      6 => lease = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { key, create_revision, mod_revision, version, value, lease }
+}
+
+///|
+/// A `RangeRequest` (`etcdserverpb`): read the key at `key`, or the half-open range
+/// `[key, range_end)` when `range_end` is set, up to `limit` results (0 = no limit).
+pub(all) struct EtcdRangeRequest {
+  key : Bytes
+  range_end : Bytes
+  limit : Int64
+} derive(Eq)
+
+///|
+/// Encode a `RangeRequest` (key=1, range_end=2, limit=3).
+pub fn EtcdRangeRequest::encode(self : EtcdRangeRequest) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.key.length() > 0 {
+    w.bytes_(1, self.key)
+  }
+  if self.range_end.length() > 0 {
+    w.bytes_(2, self.range_end)
+  }
+  if self.limit != 0 {
+    w.int64(3, self.limit)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `RangeRequest`.
+pub fn EtcdRangeRequest::decode(
+  data : Bytes,
+) -> EtcdRangeRequest raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut key = b""
+  let mut range_end = b""
+  let mut limit = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      1 => key = r.read_bytes()
+      2 => range_end = r.read_bytes()
+      3 => limit = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { key, range_end, limit }
+}
+
+///|
+/// A `RangeResponse`: the matched key/values and the total `count` in the range (which
+/// may exceed the returned `kvs` when a `limit` capped them).
+pub(all) struct EtcdRangeResponse {
+  kvs : Array[EtcdKeyValue]
+  count : Int64
+} derive(Eq)
+
+///|
+/// Encode a `RangeResponse` (kvs=2 repeated, count=4).
+pub fn EtcdRangeResponse::encode(self : EtcdRangeResponse) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  for kv in self.kvs {
+    w.message_(2, kv.encode())
+  }
+  if self.count != 0 {
+    w.int64(4, self.count)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `RangeResponse`; each `kvs` entry is a nested `KeyValue` message.
+pub fn EtcdRangeResponse::decode(
+  data : Bytes,
+) -> EtcdRangeResponse raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let kvs : Array[EtcdKeyValue] = []
+  let mut count = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      2 => kvs.push(EtcdKeyValue::decode(r.read_bytes()))
+      4 => count = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { kvs, count }
+}
+
+///|
+/// A `PutRequest`: store `value` at `key`, optionally under `lease`.
+pub(all) struct EtcdPutRequest {
+  key : Bytes
+  value : Bytes
+  lease : Int64
+} derive(Eq)
+
+///|
+/// Encode a `PutRequest` (key=1, value=2, lease=3).
+pub fn EtcdPutRequest::encode(self : EtcdPutRequest) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.key.length() > 0 {
+    w.bytes_(1, self.key)
+  }
+  if self.value.length() > 0 {
+    w.bytes_(2, self.value)
+  }
+  if self.lease != 0 {
+    w.int64(3, self.lease)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `PutRequest`.
+pub fn EtcdPutRequest::decode(
+  data : Bytes,
+) -> EtcdPutRequest raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut key = b""
+  let mut value = b""
+  let mut lease = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      1 => key = r.read_bytes()
+      2 => value = r.read_bytes()
+      3 => lease = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { key, value, lease }
+}

etcd_wbtest.mbt

@@ -0,0 +1,84 @@
+// Reference vectors produced by CPython's `protobuf` (6.33) compiling the etcd
+// `etcdserverpb` KV messages, so the self-built encoders must match the exact bytes a
+// real etcd server produces and reads.
+
+///|
+fn etcd_nib(c : Char) -> Int {
+  if c >= '0' && c <= '9' {
+    c.to_int() - '0'.to_int()
+  } else if c >= 'a' && c <= 'f' {
+    c.to_int() - 'a'.to_int() + 10
+  } else {
+    c.to_int() - 'A'.to_int() + 10
+  }
+}
+
+///|
+fn etcd_unhex(s : String) -> Bytes {
+  let buf = Buffer()
+  let chars = s.to_array()
+  let mut i = 0
+  while i + 1 < chars.length() {
+    buf.write_byte(
+      ((etcd_nib(chars[i]) << 4) | etcd_nib(chars[i + 1])).to_byte(),
+    )
+    i += 2
+  }
+  buf.to_bytes()
+}
+
+///|
+test "etcd RangeRequest encodes to the exact protobuf bytes protobuf produces" {
+  let req : EtcdRangeRequest = { key: b"foo", range_end: b"fop", limit: 100 }
+  // protobuf: RangeRequest(key="foo", range_end="fop", limit=100)
+  assert_eq(req.encode() == etcd_unhex("0a03666f6f1203666f701864"), true)
+  assert_eq(EtcdRangeRequest::decode(req.encode()) == req, true)
+}
+
+///|
+test "etcd KeyValue encodes to the exact protobuf bytes and round-trips" {
+  let kv : EtcdKeyValue = {
+    key: b"k",
+    create_revision: 2,
+    mod_revision: 3,
+    version: 1,
+    value: b"v",
+    lease: 7,
+  }
+  // protobuf: KeyValue(key="k", create_revision=2, mod_revision=3, version=1,
+  // value="v", lease=7)
+  assert_eq(kv.encode() == etcd_unhex("0a016b1002180320012a01763007"), true)
+  assert_eq(EtcdKeyValue::decode(kv.encode()) == kv, true)
+}
+
+///|
+test "etcd PutRequest encodes to the exact protobuf bytes and round-trips" {
+  let put : EtcdPutRequest = { key: b"k", value: b"v", lease: 7 }
+  // protobuf: PutRequest(key="k", value="v", lease=7)
+  assert_eq(put.encode() == etcd_unhex("0a016b1201761807"), true)
+  assert_eq(EtcdPutRequest::decode(put.encode()) == put, true)
+}
+
+///|
+test "etcd RangeResponse carries nested KeyValues and a count through a round-trip" {
+  let resp : EtcdRangeResponse = {
+    kvs: [
+      { ..EtcdKeyValue::empty(), key: b"a", value: b"1", mod_revision: 5 },
+      { ..EtcdKeyValue::empty(), key: b"b", value: b"2", mod_revision: 6 },
+    ],
+    count: 2,
+  }
+  let decoded = EtcdRangeResponse::decode(resp.encode())
+  assert_eq(decoded.count, 2)
+  assert_eq(decoded.kvs.length(), 2)
+  assert_eq(decoded.kvs[0].key == b"a", true)
+  assert_eq(decoded.kvs[1].mod_revision, 6)
+  assert_eq(decoded == resp, true)
+}
+
+///|
+test "etcd default (empty/zero) fields are omitted, per proto3" {
+  // A RangeRequest for a single key with no range_end or limit encodes only field 1.
+  let req : EtcdRangeRequest = { key: b"x", range_end: b"", limit: 0 }
+  assert_eq(req.encode() == etcd_unhex("0a0178"), true)
+}

moon.mod

@@ -22,6 +22,6 @@ description = "moonzero — a service framework for MoonBit (← go-zero): confi
 import {
   "Lfan-ke/moonapi@0.6.2",
   "Lfan-ke/moonasgi@0.6.1",
-  "Lfan-ke/moonrpc@0.5.0",
+  "Lfan-ke/moonrpc@0.7.0",
   "moonbitlang/async@0.20.3",
 }