feat(moonzero): real HTTP-over-TCP consul socket client with a live-consul CI job.

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

Author: Lfan-ke

.github/workflows/ci.yml

@@ -87,3 +87,47 @@ jobs:
           MOON_REDIS_TEST: "1"
           MOON_REDIS_HOST: "127.0.0.1"
         run: moon test --target native
+
+  # consul's agent needs `agent -dev` arguments, so it is started with docker run
+  # rather than a service container; the native discov tests register a service with a
+  # TTL check, pass it, resolve the healthy set, and deregister, gated on
+  # MOON_CONSUL_TEST.
+  discovery-consul:
+    name: discovery drivers (native, live consul)
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - name: install
+        run: |
+          curl -fsSL https://cli.moonbitlang.com/install/unix.sh | bash
+          echo "$HOME/.moon/bin" >> $GITHUB_PATH
+
+      - name: post install
+        run: |
+          moon version --all
+          moon update
+
+      - name: start consul (dev agent)
+        run: |
+          docker run -d --name consul -p 8500:8500 hashicorp/consul:1.19 agent -dev -client 0.0.0.0
+
+      - name: wait for consul
+        run: |
+          for i in $(seq 1 30); do
+            if curl -sf http://127.0.0.1:8500/v1/status/leader | grep -qE '[0-9]'; then
+              echo "consul is up"; break
+            fi
+            echo "waiting for consul... ($i)"; sleep 2
+          done
+          curl -sf http://127.0.0.1:8500/v1/agent/self | head -c 200
+
+      - name: discovery round trip against live consul
+        env:
+          MOON_CONSUL_TEST: "1"
+          MOON_CONSUL_HOST: "127.0.0.1"
+        run: moon test --target native

consul.mbt

@@ -25,6 +25,18 @@ pub fn ConsulResponse::new(status : Int, body : Bytes) -> ConsulResponse {
   { status, body }
 }
 
+///|
+/// The HTTP status code.
+pub fn ConsulResponse::status(self : ConsulResponse) -> Int {
+  self.status
+}
+
+///|
+/// The raw response body bytes.
+pub fn ConsulResponse::body(self : ConsulResponse) -> Bytes {
+  self.body
+}
+
 ///|
 /// A transport to a consul agent: it performs one HTTP request (`method` and `path`,
 /// with `body` for writes) and returns the response. Implementations own the medium —
@@ -90,6 +102,24 @@ pub fn ConsulClient::register_service(
   port : Int,
   ttl_secs : Int,
 ) -> Unit raise ConsulError {
+  let _ = self.call(
+    "PUT",
+    "/v1/agent/service/register",
+    consul_register_body(id, name, address, port, ttl_secs),
+  )
+}
+
+///|
+/// The JSON body of a `service/register` request: the instance's id, name, address, and
+/// port, plus a TTL check that consul deregisters `ttl*3` seconds after it stops
+/// passing. Exposed so the native HTTP-socket path builds the exact same body.
+pub fn consul_register_body(
+  id : String,
+  name : String,
+  address : String,
+  port : Int,
+  ttl_secs : Int,
+) -> Bytes {
   let check : Map[String, Json] = Map([
     ("TTL", (ttl_secs.to_string() + "s").to_json()),
     (
@@ -104,11 +134,7 @@ pub fn ConsulClient::register_service(
     ("Port", port.to_json()),
     ("Check", check.to_json()),
   ])
-  let _ = self.call(
-    "PUT",
-    "/v1/agent/service/register",
-    @utf8.encode(body.to_json().stringify()),
-  )
+  @utf8.encode(body.to_json().stringify())
 }
 
 ///|
@@ -177,6 +203,13 @@ pub fn ConsulClient::agent_services(
   out
 }
 
+///|
+/// Parse a consul `health/service` JSON body into endpoints — exposed so the native
+/// HTTP-socket path decodes a real agent's response exactly as the client does.
+pub fn consul_parse_health(body : Bytes) -> Array[Endpoint] raise ConsulError {
+  parse_health_response(body)
+}
+
 ///|
 /// Parse a consul `health/service` JSON array into endpoints.
 fn parse_health_response(body : Bytes) -> Array[Endpoint] raise ConsulError {

discov/consul_socket.mbt

@@ -0,0 +1,56 @@
+// HTTP over a real TCP socket: a consul agent transport the discovery flow drives
+// against a live consul. moonzero's root `ConsulHttp` is synchronous (so the driver and
+// its mock run on every backend), but socket I/O is async, so the real transport is a
+// separate native-only path here — it reuses the root's pure HTTP/1 codec
+// (`http1_request` to frame the request, `http1_parse_response` to parse the reply) and
+// only adds the async connect/send/receive. Native-only, because `moonbitlang/async`
+// has no JS/Wasm backend.
+
+///|
+/// A consul agent reachable over HTTP at `host:port`. Each request opens a fresh
+/// connection (HTTP/1.0 with `Connection: close`), the simplest correct model for the
+/// agent's short JSON round trips.
+pub struct ConsulSocket {
+  host : String
+  port : Int
+}
+
+///|
+/// A transport to the consul agent at `host:port` (default port 8500).
+pub fn ConsulSocket::new(host : String, port? : Int = 8500) -> ConsulSocket {
+  { host, port }
+}
+
+///|
+/// Perform one HTTP request against the agent and return its status and body.
+pub async fn ConsulSocket::request(
+  self : ConsulSocket,
+  verb : String,
+  path : String,
+  body : Bytes,
+) -> @moonzero.ConsulResponse {
+  let addr = @socket.Addr::resolve(self.host, port=self.port)
+  let tcp = @socket.Tcp::connect(addr)
+  let content_type = if body.length() > 0 { "application/json" } else { "" }
+  let host_header = self.host + ":" + self.port.to_string()
+  let request = @moonzero.http1_request(
+    verb,
+    path,
+    host_header,
+    body,
+    content_type~,
+  )
+  let response = try {
+    (tcp : &@io.Writer).write(request)
+    let raw = (tcp : &@io.Reader).read_all()
+    let parsed = @moonzero.http1_parse_response(raw.binary())
+    @moonzero.ConsulResponse::new(parsed.status(), parsed.body())
+  } catch {
+    e => {
+      tcp.close()
+      raise e
+    }
+  }
+  tcp.close()
+  response
+}

discov/consul_socket_test.mbt

@@ -0,0 +1,63 @@
+// Integration test against a REAL consul agent. Gated on `MOON_CONSUL_TEST`, so it is
+// skipped in the normal suite and runs only in the CI job that stands up a consul
+// container. It drives the exact discovery wire — register a service with a TTL check,
+// pass the check, resolve the healthy set, deregister — over the real HTTP socket,
+// reusing the root's request-body builder and health parser, so a green run proves the
+// HTTP/1 codec and the discovery flow against a live consul.
+
+///|
+/// The consul host for the integration test: `MOON_CONSUL_HOST`, or localhost.
+fn consul_test_host() -> String {
+  match @env.get_env_var("MOON_CONSUL_HOST") {
+    Some(h) => h
+    None => "127.0.0.1"
+  }
+}
+
+///|
+/// Register `endpoint` for `service` and pass its TTL check so it is healthy at once.
+async fn consul_put_healthy(
+  sock : ConsulSocket,
+  service : String,
+  id : String,
+  host : String,
+  port : Int,
+) -> Unit {
+  let ok = sock.request(
+    "PUT",
+    "/v1/agent/service/register",
+    @moonzero.consul_register_body(id, service, host, port, 30),
+  )
+  assert_eq(ok.status(), 200)
+  let passed = sock.request("PUT", "/v1/agent/check/pass/service:" + id, b"")
+  assert_eq(passed.status(), 200)
+}
+
+///|
+async test "integration: consul discovery register/resolve/deregister over a real consul" {
+  guard @env.get_env_var("MOON_CONSUL_TEST") is Some(_) else { return }
+  let sock = ConsulSocket::new(consul_test_host())
+  let id1 = "greeter-10.0.0.1-9090"
+  let id2 = "greeter-10.0.0.2-9090"
+  // Clean any leftover from a prior run.
+  let _ = sock.request("PUT", "/v1/agent/service/deregister/" + id1, b"")
+  let _ = sock.request("PUT", "/v1/agent/service/deregister/" + id2, b"")
+  // Register two healthy instances.
+  consul_put_healthy(sock, "greeter", id1, "10.0.0.1", 9090)
+  consul_put_healthy(sock, "greeter", id2, "10.0.0.2", 9090)
+  // Resolve the healthy set.
+  let resp = sock.request("GET", "/v1/health/service/greeter?passing=true", b"")
+  assert_eq(resp.status(), 200)
+  let eps = @moonzero.consul_parse_health(resp.body())
+  assert_eq(eps.length(), 2)
+  let addrs = eps.map(e => e.address())
+  assert_eq(addrs.contains("10.0.0.1:9090"), true)
+  assert_eq(addrs.contains("10.0.0.2:9090"), true)
+  // Deregister both; the healthy set is then empty.
+  let _ = sock.request("PUT", "/v1/agent/service/deregister/" + id1, b"")
+  let _ = sock.request("PUT", "/v1/agent/service/deregister/" + id2, b"")
+  let after = sock.request(
+    "GET", "/v1/health/service/greeter?passing=true", b"",
+  )
+  assert_eq(@moonzero.consul_parse_health(after.body()).length(), 0)
+}