feat(moonzero): etcd Lease and watch Event protobuf messages.

The discovery path needs a lease for instance liveness and watch events for change notification. LeaseGrantRequest/Response, LeaseKeepAliveRequest/Response, and the mvccpb Event (PUT/DELETE with its KeyValue) now encode and decode over moonrpc's protobuf runtime with etcd's rpc.proto field numbers. Verified against reference vectors CPython's protobuf produced (mutation-checked on the proto3 default-omission of a PUT event's type) and round-tripped on all four backends.

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

Author: Lfan-ke

etcd.mbt

@@ -215,3 +215,232 @@ pub fn EtcdPutRequest::decode(
   }
   { key, value, lease }
 }
+
+///|
+/// A `LeaseGrantRequest`: ask etcd for a lease living `ttl` seconds (`id` 0 lets the
+/// server assign one). A registered service key attaches to the lease and vanishes
+/// when the lease expires — go-zero's instance liveness mechanism.
+pub(all) struct EtcdLeaseGrantRequest {
+  ttl : Int64
+  id : Int64
+} derive(Eq)
+
+///|
+/// Encode a `LeaseGrantRequest` (TTL=1, ID=2).
+pub fn EtcdLeaseGrantRequest::encode(self : EtcdLeaseGrantRequest) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.ttl != 0 {
+    w.int64(1, self.ttl)
+  }
+  if self.id != 0 {
+    w.int64(2, self.id)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `LeaseGrantRequest`.
+pub fn EtcdLeaseGrantRequest::decode(
+  data : Bytes,
+) -> EtcdLeaseGrantRequest raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut ttl = 0L
+  let mut id = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      1 => ttl = r.read_int64()
+      2 => id = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { ttl, id }
+}
+
+///|
+/// A `LeaseGrantResponse`: the granted lease `id`, its actual `ttl`, and an `error`
+/// string when the grant failed.
+pub(all) struct EtcdLeaseGrantResponse {
+  id : Int64
+  ttl : Int64
+  error : String
+} derive(Eq)
+
+///|
+/// Encode a `LeaseGrantResponse` (ID=2, TTL=3, error=4).
+pub fn EtcdLeaseGrantResponse::encode(self : EtcdLeaseGrantResponse) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.id != 0 {
+    w.int64(2, self.id)
+  }
+  if self.ttl != 0 {
+    w.int64(3, self.ttl)
+  }
+  if self.error.length() > 0 {
+    w.string_(4, self.error)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `LeaseGrantResponse`.
+pub fn EtcdLeaseGrantResponse::decode(
+  data : Bytes,
+) -> EtcdLeaseGrantResponse raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut id = 0L
+  let mut ttl = 0L
+  let mut error = ""
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      2 => id = r.read_int64()
+      3 => ttl = r.read_int64()
+      4 => error = r.read_string()
+      _ => r.skip(wire)
+    }
+  }
+  { id, ttl, error }
+}
+
+///|
+/// A `LeaseKeepAliveRequest`: renew lease `id` before it expires. A discovery client
+/// streams these to keep its instance registered.
+pub(all) struct EtcdLeaseKeepAliveRequest {
+  id : Int64
+} derive(Eq)
+
+///|
+/// Encode a `LeaseKeepAliveRequest` (ID=1).
+pub fn EtcdLeaseKeepAliveRequest::encode(
+  self : EtcdLeaseKeepAliveRequest,
+) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.id != 0 {
+    w.int64(1, self.id)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `LeaseKeepAliveRequest`.
+pub fn EtcdLeaseKeepAliveRequest::decode(
+  data : Bytes,
+) -> EtcdLeaseKeepAliveRequest raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut id = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      1 => id = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { id, }
+}
+
+///|
+/// A `LeaseKeepAliveResponse`: the renewed lease `id` and its remaining `ttl` (0 =
+/// the lease has expired).
+pub(all) struct EtcdLeaseKeepAliveResponse {
+  id : Int64
+  ttl : Int64
+} derive(Eq)
+
+///|
+/// Encode a `LeaseKeepAliveResponse` (ID=2, TTL=3).
+pub fn EtcdLeaseKeepAliveResponse::encode(
+  self : EtcdLeaseKeepAliveResponse,
+) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.id != 0 {
+    w.int64(2, self.id)
+  }
+  if self.ttl != 0 {
+    w.int64(3, self.ttl)
+  }
+  w.to_bytes()
+}
+
+///|
+/// Decode a `LeaseKeepAliveResponse`.
+pub fn EtcdLeaseKeepAliveResponse::decode(
+  data : Bytes,
+) -> EtcdLeaseKeepAliveResponse raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut id = 0L
+  let mut ttl = 0L
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      2 => id = r.read_int64()
+      3 => ttl = r.read_int64()
+      _ => r.skip(wire)
+    }
+  }
+  { id, ttl }
+}
+
+///|
+/// The kind of change a watch `Event` reports (`mvccpb.Event.EventType`): a key was
+/// `Put` (created or updated) or `Delete`d.
+pub(all) enum EtcdEventType {
+  Put
+  Delete
+} derive(Eq)
+
+///|
+/// The protobuf enum number of an event type (`PUT` = 0, `DELETE` = 1).
+pub fn EtcdEventType::to_int(self : EtcdEventType) -> Int {
+  match self {
+    Put => 0
+    Delete => 1
+  }
+}
+
+///|
+/// The event type for a protobuf enum number; unknown numbers read as `Put`.
+pub fn EtcdEventType::from_int(n : Int) -> EtcdEventType {
+  if n == 1 {
+    Delete
+  } else {
+    Put
+  }
+}
+
+///|
+/// A watch `Event` (`mvccpb.Event`): a change to one key, carrying the resulting
+/// `KeyValue` (for a delete, the key with cleared metadata). This is what a discovery
+/// watcher folds into add/remove of a service instance.
+pub(all) struct EtcdEvent {
+  event_type : EtcdEventType
+  kv : EtcdKeyValue
+} derive(Eq)
+
+///|
+/// Encode an `Event` (type=1, kv=2). `PUT` (0) is the proto3 default and omitted.
+pub fn EtcdEvent::encode(self : EtcdEvent) -> Bytes {
+  let w = @moonrpc.PbWriter::new()
+  if self.event_type.to_int() != 0 {
+    w.int32(1, self.event_type.to_int())
+  }
+  w.message_(2, self.kv.encode())
+  w.to_bytes()
+}
+
+///|
+/// Decode an `Event`.
+pub fn EtcdEvent::decode(data : Bytes) -> EtcdEvent raise @moonrpc.PbError {
+  let r = @moonrpc.PbReader::new(data)
+  let mut event_type = EtcdEventType::Put
+  let mut kv = EtcdKeyValue::empty()
+  while !r.eof() {
+    let (field, wire) = r.read_tag()
+    match field {
+      1 => event_type = EtcdEventType::from_int(r.read_int32())
+      2 => kv = EtcdKeyValue::decode(r.read_bytes())
+      _ => r.skip(wire)
+    }
+  }
+  { event_type, kv }
+}

etcd_wbtest.mbt

@@ -82,3 +82,48 @@ test "etcd default (empty/zero) fields are omitted, per proto3" {
   let req : EtcdRangeRequest = { key: b"x", range_end: b"", limit: 0 }
   assert_eq(req.encode() == etcd_unhex("0a0178"), true)
 }
+
+///|
+test "etcd Lease messages encode to the exact protobuf bytes and round-trip" {
+  let grant : EtcdLeaseGrantRequest = { ttl: 60, id: 1234 }
+  assert_eq(grant.encode() == etcd_unhex("083c10d209"), true)
+  assert_eq(EtcdLeaseGrantRequest::decode(grant.encode()) == grant, true)
+  let granted : EtcdLeaseGrantResponse = { id: 1234, ttl: 60, error: "" }
+  assert_eq(granted.encode() == etcd_unhex("10d209183c"), true)
+  assert_eq(EtcdLeaseGrantResponse::decode(granted.encode()) == granted, true)
+  let ka : EtcdLeaseKeepAliveRequest = { id: 1234 }
+  assert_eq(ka.encode() == etcd_unhex("08d209"), true)
+  assert_eq(EtcdLeaseKeepAliveRequest::decode(ka.encode()) == ka, true)
+  let kar : EtcdLeaseKeepAliveResponse = { id: 1234, ttl: 59 }
+  assert_eq(kar.encode() == etcd_unhex("10d209183b"), true)
+  assert_eq(EtcdLeaseKeepAliveResponse::decode(kar.encode()) == kar, true)
+}
+
+///|
+test "etcd watch Event encodes to the exact protobuf bytes, PUT vs DELETE" {
+  // A PUT event carries the full key/value; type PUT (0) is the proto3 default so it
+  // is omitted from the wire.
+  let put : EtcdEvent = {
+    event_type: Put,
+    kv: {
+      ..EtcdKeyValue::empty(),
+      key: b"svc/a",
+      value: b"1.2.3.4",
+      mod_revision: 9,
+    },
+  }
+  assert_eq(
+    put.encode() == etcd_unhex("12120a057376632f6118092a07312e322e332e34"),
+    true,
+  )
+  assert_eq(EtcdEvent::decode(put.encode()) == put, true)
+  // A DELETE event writes the type explicitly and just the key.
+  let del : EtcdEvent = {
+    event_type: Delete,
+    kv: { ..EtcdKeyValue::empty(), key: b"svc/a" },
+  }
+  assert_eq(del.encode() == etcd_unhex("080112070a057376632f61"), true)
+  let decoded = EtcdEvent::decode(del.encode())
+  assert_eq(decoded.event_type == Delete, true)
+  assert_eq(decoded.kv.key == b"svc/a", true)
+}