feat(moonzero): etcd Watch stream over a real socket — subscribe and observe live key events.

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

Author: Lfan-ke

discov/etcd_socket.mbt

@@ -83,6 +83,29 @@ pub async fn EtcdSocket::delete_range(
   )
 }
 
+///|
+/// `Watch.Watch`: subscribe to changes on the key or range in `req` and read the first
+/// `count` `WatchResponse`s off the stream — the `created` acknowledgement, then events as
+/// the watched keys change. The responses are read inline (no background reader), so the
+/// caller collects what it needs and closes the connection; a live subscriber loops calling
+/// this. Changes are driven elsewhere (a `Put` on another connection), the way go-zero's
+/// discov subscriber watches etcd while publishers register.
+pub async fn EtcdSocket::watch(
+  self : EtcdSocket,
+  req : @moonzero.EtcdWatchCreateRequest,
+  count : Int,
+) -> Array[@moonzero.EtcdWatchResponse] {
+  let request = @moonzero.EtcdWatchRequest::Create(req).encode()
+  let raw = self.channel.stream_take(
+    "/etcdserverpb.Watch/Watch", request, count,
+  )
+  let out : Array[@moonzero.EtcdWatchResponse] = []
+  for bytes in raw {
+    out.push(@moonzero.EtcdWatchResponse::decode(bytes))
+  }
+  out
+}
+
 ///|
 /// `Lease.LeaseGrant`: obtain a lease with the requested TTL.
 pub async fn EtcdSocket::lease_grant(

discov/etcd_socket_test.mbt

@@ -64,3 +64,47 @@ async test "integration: etcd discovery register/resolve/deregister over a real
   let after = sock.range({ key: prefix_bytes, range_end, limit: 0 })
   assert_eq(after.kvs.length(), 0)
 }
+
+///|
+async test "integration: etcd watch observes a live PUT over a real watch stream" {
+  guard @env.get_env_var("MOON_ETCD_TEST") is Some(_) else { return }
+  @async.with_task_group(g => {
+    let key = @utf8.encode("moonzero-ci/watch/probe")
+    // Clean any leftover on a setup connection.
+    let setup = EtcdSocket::connect(etcd_test_host(), 2379)
+    let _ = setup.delete_range({ key, range_end: b"", prev_kv: false })
+    setup.close()
+    // A concurrent publisher Puts the key once the watch has had time to establish.
+    let putter = g.spawn(() => {
+      @async.sleep(500)
+      let pub_conn = EtcdSocket::connect(etcd_test_host(), 2379)
+      let _ = pub_conn.put({
+        key,
+        value: @utf8.encode("10.0.0.1:9090"),
+        lease: 0L,
+      })
+      pub_conn.close()
+    })
+    // Watch and read the created ack plus the PUT event the publisher triggers.
+    let sock = EtcdSocket::connect(etcd_test_host(), 2379)
+    let responses = sock.watch({ key, range_end: b"", start_revision: 0L }, 2)
+    sock.close()
+    let _ = putter.wait()
+    assert_eq(responses.length() >= 2, true)
+    assert_eq(responses[0].created, true)
+    let mut saw_put = false
+    for r in responses {
+      for e in r.events {
+        if e.event_type == @moonzero.EtcdEventType::Put &&
+          e.kv.value == @utf8.encode("10.0.0.1:9090") {
+          saw_put = true
+        }
+      }
+    }
+    assert_eq(saw_put, true)
+    // Cleanup.
+    let cleanup = EtcdSocket::connect(etcd_test_host(), 2379)
+    let _ = cleanup.delete_range({ key, range_end: b"", prev_kv: false })
+    cleanup.close()
+  })
+}

discov/etcd_watch_test.mbt

@@ -0,0 +1,54 @@
+// The etcd Watch stream, driven against a mock etcd Watch RPC over a real socket: the
+// server streams a `created` acknowledgement then a PUT event, and `EtcdSocket::watch` reads
+// and decodes both off the wire. The real-etcd counterpart is gated in the integration test.
+
+///|
+async test "etcd watch: reads created then a PUT event off a streamed Watch RPC" {
+  @async.with_task_group(g => {
+    let created : @moonzero.EtcdWatchResponse = {
+      watch_id: 7L,
+      created: true,
+      canceled: false,
+      events: [],
+    }
+    let event : @moonzero.EtcdWatchResponse = {
+      watch_id: 7L,
+      created: false,
+      canceled: false,
+      events: [
+        {
+          event_type: @moonzero.EtcdEventType::Put,
+          kv: {
+            key: b"svc/1",
+            create_revision: 2L,
+            mod_revision: 3L,
+            version: 1L,
+            value: b"10.0.0.1:9090",
+            lease: 0L,
+          },
+        },
+      ],
+    }
+    let server = @net.GrpcServer::new()
+    server.register_server_streaming("/etcdserverpb.Watch/Watch", (_ctx, _req) => {
+      [created.encode(), event.encode()]
+    })
+    let ts = g.spawn(() => server.serve(port=18140))
+    @async.sleep(300)
+    let sock = EtcdSocket::connect("127.0.0.1", 18140)
+    let responses = sock.watch(
+      { key: b"svc/", range_end: b"svc0", start_revision: 0L },
+      2,
+    )
+    sock.close()
+    ts.cancel()
+    assert_eq(responses.length(), 2)
+    assert_eq(responses[0].created, true)
+    assert_eq(responses[1].events.length(), 1)
+    assert_eq(
+      responses[1].events[0].event_type == @moonzero.EtcdEventType::Put,
+      true,
+    )
+    assert_eq(responses[1].events[0].kv.value == b"10.0.0.1:9090", 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.8.0",
+  "Lfan-ke/moonrpc@0.8.1",
   "moonbitlang/async@0.20.3",
 }