feat(dchain): expose native CLOB order ingestion over luxd HTTP router

The in-luxd native D-Chain VM (pkg/dchain) bootstrapped consensus but never exposed order ingestion: VM.CreateHandlers returned an empty map, so luxd mounted nothing and /ext/bc/D/* 404'd. RegisterCLOB (the clob_* -> submitTx -> mempool -> consensus -> Verify-match path) was only reachable over the ZAP socket transport, which only cmd/dvenue (the abandoned standalone venue) wires. Wire ingestion the idiomatic luxd way: CreateHandlers now returns one HTTP handler per CLOB method (/dex/clob_place, /dex/clob_submit, ...), each adapting the EXACT same rpc.RawHandler RegisterCLOB binds (one method->handler table, two transports — vm.clobMethods). luxd mounts each as an exact route under /ext/bc/<chainID> (and the D alias) through the node's own router. An order POSTed to /ext/bc/D/dex/clob_submit flows submitTx -> mempool.Add -> consensus PendingTxs -> BuildBlock -> Block.Verify (the matcher) -> Block.Accept, and the response is the CONSENSUS-COMPUTED outcome. No external matcher, no relay, no side port, no node image change — the chain is the matcher. TDD: TestIngestCrossingOrdersMatchInBlock proves two crossing orders submitted over the HTTP handler produce a fill committed as D-Chain state (trade: row, height advances). CGO=0; full pkg/dchain suite green; plugin cross-builds linux/amd64.
2026-09-24 00:26:31 +00:00 · 2026-06-22 11:03:15 -07:00
parent 59c9a6c347
commit 80e1018781
4 changed files with 397 additions and 14 deletions
@@ -31,15 +31,22 @@ import (
 // identical constants (they cannot import this cgo-tagged package), so a frame
 // built by any of them is byte-identical to what this server consumes/produces.

-// RegisterCLOB wires the four clob_* handlers onto an rpc.Server. It is the
-// integration seam the venue entrypoint (cmd/dchain) and tests use to expose the
-// d-chain matcher over ZAP. Additive: the caller may register other methods on
-// the same server.
-func (vm *VM) RegisterCLOB(server rpc.Server) error {
-	regs := []struct {
-		method  string
-		handler rpc.RawHandler
-	}{
+// clobMethod pairs a frozen ZAP method name with its raw handler. The handler
+// signature is github.com/luxfi/rpc.RawHandler (func(ctx, payload) (resp, err)) —
+// the SAME func value whether it is reached over the ZAP socket transport
+// (RegisterCLOB) or over the node's HTTP router (CreateHandlers / ingest.go). One
+// table, two transports: the ingestion seam never duplicates the handler set.
+type clobMethod struct {
+	method  string
+	handler rpc.RawHandler
+}
+
+// clobMethods is the single authoritative method->handler table for the CLOB
+// surface. Both transports fold over it: RegisterCLOB registers each on a ZAP
+// rpc.Server; the HTTP mux (ingest.go) routes POST .../<method> to the same
+// handler. A method added here is reachable over both transports automatically.
+func (vm *VM) clobMethods() []clobMethod {
+	return []clobMethod{
 		{zapwire.MethodEnsureMarket, vm.handleEnsureMarket},
 		{zapwire.MethodPlace, vm.handlePlace},
 		{zapwire.MethodCancel, vm.handleCancel},
@@ -48,9 +55,17 @@ func (vm *VM) RegisterCLOB(server rpc.Server) error {
 		{zapwire.MethodDeposit, vm.handleDeposit},
 		{zapwire.MethodWithdraw, vm.handleWithdraw},
 	}
-	for _, r := range regs {
-		if err := server.RegisterRaw(r.method, r.handler); err != nil {
-			return fmt.Errorf("dchain: register %s: %w", r.method, err)
+}
+
+// RegisterCLOB wires the clob_* handlers onto an rpc.Server (the ZAP socket
+// transport). It is the seam the standalone venue entrypoint (cmd/dvenue) and the
+// socket-level tests use. The in-luxd plugin does NOT use this path — it exposes
+// the SAME handlers over the node's HTTP router via CreateHandlers (ingest.go).
+// Additive: the caller may register other methods on the same server.
+func (vm *VM) RegisterCLOB(server rpc.Server) error {
+	for _, m := range vm.clobMethods() {
+		if err := server.RegisterRaw(m.method, m.handler); err != nil {
+			return fmt.Errorf("dchain: register %s: %w", m.method, err)
 		}
 	}
 	return nil
@@ -0,0 +1,101 @@
+// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package dchain
+
+import (
+	"context"
+	"io"
+	"net/http"
+
+	"github.com/luxfi/dex/pkg/zapwire"
+	"github.com/luxfi/rpc"
+)
+
+// ingest.go is the D-Chain CLOB ingestion seam for the IN-LUXD native VM. When
+// the VM runs as a luxd plugin (cmd/dchain via github.com/luxfi/vm/rpc.Serve),
+// luxd asks the VM for HTTP handlers (CreateHandlers); the plugin transport stands
+// up a local http.Server per handler inside the plugin process and the node
+// reverse-proxies to it. luxd mounts each handler at /ext/bc/<chainID-or-alias>
+// + <endpoint-key> as an EXACT gorilla/mux route (node/server/http/router.go:
+// `r.router.Handle(base+endpoint, handler)` — exact path, no subtree). So we
+// return ONE handler per method, each keyed by its full sub-path:
+//
+//	/ext/bc/D/dex/clob_place    -> handlePlace
+//	/ext/bc/D/dex/clob_submit   -> handleSubmit
+//	/ext/bc/D/dex/clob_cancel   -> handleCancel
+//	...
+//
+// The request body is the FROZEN zapwire payload verbatim; the response body is
+// the handler's raw bytes (zapwire ack / fills / balance frame).
+//
+// THE INVARIANT THAT MATTERS: this is pure transport. The clob_* handler each
+// route dispatches to is the EXACT func value RegisterCLOB binds on the ZAP socket
+// (vm.clobMethods) — so an order POSTed here flows submitTx -> mempool.Add ->
+// consensus PendingTxs -> BuildBlock -> Block.Verify (the matcher) -> Block.Accept,
+// and the bytes returned are the CONSENSUS-COMPUTED outcome (every validator
+// re-derived the same fills at Verify). It NEVER relays to an external matcher and
+// NEVER mutates the book synchronously: the chain is the matcher.
+
+// ingestNamespace is the path segment grouping the CLOB methods under the chain
+// route. The full external path of a method m is /ext/bc/<chainID>/dex/<m>.
+const ingestNamespace = "dex"
+
+// maxIngestBody bounds a request body. The largest frozen frame is the 96-byte
+// OpenMarket; 4 KiB is comfortably above every frame and rejects a runaway body
+// without parsing it (defence at the boundary — the handlers also length-check).
+const maxIngestBody = 4 << 10
+
+// httpHandlers returns the VM's CLOB methods as luxd HTTP handlers, keyed by their
+// full endpoint sub-path ("/dex/<method>"). It folds the SAME vm.clobMethods table
+// the ZAP socket transport uses — one method->handler table, two transports — so a
+// method added to clobMethods is reachable over HTTP automatically. luxd registers
+// each key as an exact route under /ext/bc/<chainID> (see file doc).
+func (vm *VM) httpHandlers() map[string]http.Handler {
+	out := make(map[string]http.Handler, len(vm.clobMethods()))
+	for _, m := range vm.clobMethods() {
+		out["/"+ingestNamespace+"/"+m.method] = rawHandlerHTTP(m.handler)
+	}
+	return out
+}
+
+// rawHandlerHTTP adapts a github.com/luxfi/rpc.RawHandler (the ZAP handler
+// signature) to an http.Handler. It enforces POST, reads the raw frozen-frame body
+// as the payload, invokes the handler with the REQUEST context (so a client
+// disconnect cancels the consensus wait and submitTx withdraws the tx — RED #4),
+// and writes the handler's raw response bytes back. A handler error becomes 502
+// with the error text (the order could not be turned into a consensus outcome); a
+// success is application/octet-stream — the frozen wire frame, never JSON.
+func rawHandlerHTTP(h rpc.RawHandler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			w.Header().Set("Allow", http.MethodPost)
+			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+		payload, err := io.ReadAll(io.LimitReader(r.Body, maxIngestBody))
+		if err != nil {
+			http.Error(w, "read body: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+		resp, err := h(r.Context(), payload)
+		if err != nil {
+			// The order/op could not be sequenced into a consensus outcome (e.g.
+			// the client disconnected and the wait was cancelled, or the frame was
+			// undecodable). 502: the request was well-formed but the VM could not
+			// produce an upstream (consensus) result.
+			http.Error(w, err.Error(), http.StatusBadGateway)
+			return
+		}
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write(resp)
+	})
+}
+
+// compile-time guards.
+var (
+	_ rpc.RawHandler = func(context.Context, []byte) ([]byte, error) { return nil, nil }
+	// the frozen method names are what we route on.
+	_ = zapwire.MethodSubmit
+)
@@ -0,0 +1,260 @@
+// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package dchain
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/luxfi/consensus/engine/chain/block"
+	"github.com/luxfi/database/memdb"
+	"github.com/luxfi/dex/pkg/zapwire"
+	"github.com/luxfi/log"
+)
+
+// ingest_test.go proves the IN-LUXD ingestion seam (ingest.go / CreateHandlers):
+// an order submitted over the VM's HTTP handler — the EXACT handler luxd mounts at
+// /ext/bc/<D>/dex/<method> — lands in the mempool and is matched into a block by
+// the VM's own consensus lifecycle (BuildBlock -> Verify -> Accept), with the fill
+// committed as D-Chain state (a trade:<height><seq> row). The matcher is the
+// chain; there is no external venue and no relay.
+//
+// The transport under test is vm.clobHTTPHandler() driven through a real
+// httptest.Server (which is exactly what the plugin transport stands up inside the
+// plugin process). An auto-sealer goroutine plays the consensus engine, identical
+// to handler_test.go's socket harness — proving the SAME submitTx->mempool->
+// Verify-match path is reached whether the order arrives over HTTP or ZAP.
+
+// startIngestServer brings up a VM + the in-luxd HTTP CLOB handler over a real
+// httptest.Server, plus the auto-sealer. Returns the VM, the base URL
+// (.../dex), and a stop func.
+func startIngestServer(t *testing.T) (*VM, string, func()) {
+	t.Helper()
+	db := memdb.New()
+	vm := &VM{}
+	toEngine := make(chan block.Message, 64)
+	if err := vm.Initialize(context.Background(), block.Init{
+		DB:       db,
+		Log:      log.NewNoOpLogger(),
+		ToEngine: toEngine,
+	}); err != nil {
+		t.Fatalf("Initialize: %v", err)
+	}
+
+	// Model the node EXACTLY: CreateHandlers returns one handler per method keyed
+	// by its full sub-path ("/dex/<method>"); luxd registers each as an exact route
+	// under /ext/bc/<chainID>. We register each key as an exact route here, so the
+	// request paths and matching semantics match production (exact path, not a
+	// subtree — node/server/http/router.go uses gorilla Handle, not PathPrefix).
+	handlers, err := vm.CreateHandlers(context.Background())
+	if err != nil {
+		t.Fatalf("CreateHandlers: %v", err)
+	}
+	if len(handlers) == 0 {
+		t.Fatal("CreateHandlers returned no handlers")
+	}
+	root := http.NewServeMux()
+	for path, h := range handlers {
+		// gorilla matches exact paths; net/http ServeMux matches exact when the
+		// pattern has no trailing slash, which mirrors the node's exact-route
+		// behavior for these method endpoints.
+		root.Handle(path, h)
+	}
+	srv := httptest.NewServer(root)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); autoSealer(ctx, t, vm) }()
+
+	stop := func() {
+		cancel()
+		srv.Close()
+		_ = vm.Shutdown(context.Background())
+		wg.Wait()
+	}
+	return vm, srv.URL, stop
+}
+
+// methodURL builds the full ingestion URL for a method: <base>/dex/<method>.
+func methodURL(base, method string) string {
+	return base + "/" + ingestNamespace + "/" + method
+}
+
+// postFrame POSTs a frozen frame to <base>/dex/<method> and returns the raw
+// response body. It fails the test on any transport error or non-200 status.
+func postFrame(t *testing.T, base, method string, payload []byte) []byte {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, methodURL(base, method), bytes.NewReader(payload))
+	if err != nil {
+		t.Fatalf("new request %s: %v", method, err)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("POST %s: %v", method, err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("read %s response: %v", method, err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("POST %s: status %d body=%q", method, resp.StatusCode, string(body))
+	}
+	return body
+}
+
+// TestIngestCrossingOrdersMatchInBlock is the deliverable proof of the ingestion
+// seam: two crossing orders submitted through the in-luxd HTTP handler produce a
+// fill, and that fill is recorded as D-Chain state (a trade: row), matched by the
+// VM's BuildBlock->Verify->Accept — not by any external matcher.
+func TestIngestCrossingOrdersMatchInBlock(t *testing.T) {
+	vm, base, stop := startIngestServer(t)
+	defer stop()
+
+	var pool [32]byte
+	pool[0], pool[1] = 0xab, 0xcd
+
+	// 1) ensure_market over HTTP -> ack(placed).
+	ack := postFrame(t, base, zapwire.MethodEnsureMarket, zapwire.EncodeEnsureMarket(pool))
+	if _, status, err := zapwire.DecodeAck(ack); err != nil || status != zapwire.StatusPlaced {
+		t.Fatalf("ensure_market ack: status=%d err=%v", status, err)
+	}
+
+	// 2) place a resting ask @101 size 5 over HTTP -> ack(placed) with order id.
+	ack = postFrame(t, base, zapwire.MethodPlace,
+		zapwire.EncodePlace(pool, zapwire.SideSell, 101.0, 5.0, "maker"))
+	makerID, status, err := zapwire.DecodeAck(ack)
+	if err != nil || status != zapwire.StatusPlaced {
+		t.Fatalf("place ack: status=%d err=%v", status, err)
+	}
+	if makerID == 0 {
+		t.Fatal("place returned zero order id")
+	}
+
+	// 3) submit a crossing buy @101 size 3 over HTTP -> CONSENSUS-computed fills.
+	fillsResp := postFrame(t, base, zapwire.MethodSubmit,
+		zapwire.EncodeSubmit(pool, zapwire.SideBuy, false, 101.0, 3.0, "taker"))
+	fills, err := zapwire.DecodeFills(fillsResp)
+	if err != nil {
+		t.Fatalf("DecodeFills: %v", err)
+	}
+	if len(fills) != 1 {
+		t.Fatalf("expected 1 fill, got %d: %+v", len(fills), fills)
+	}
+	if fills[0].TakerSide != zapwire.SideBuy || fills[0].Size != 3.0 || fills[0].Price != 101.0 {
+		t.Fatalf("fill = %+v, want buy 3.0 @ 101.0", fills[0])
+	}
+
+	// 4) STATE PROOF: a trade:<height><seq> row exists — the fill is committed
+	// D-Chain state, written by Block.Accept (settle.go), not a synchronous book
+	// mutation. Scan the durable DB the VM committed into.
+	if n := countTradeRows(t, vm); n != 1 {
+		t.Fatalf("expected exactly 1 committed trade: row, got %d", n)
+	}
+
+	// 5) BLOCK PROOF: the chain advanced past genesis — the order was sequenced
+	// into a real block by consensus. (genesis is height 0; ensure_market, place,
+	// and submit each seal a block, so height >= 3.)
+	vm.mu.Lock()
+	h := vm.lastAcceptedHeight
+	vm.mu.Unlock()
+	if h < 3 {
+		t.Fatalf("lastAcceptedHeight = %d, want >= 3 (orders must seal blocks)", h)
+	}
+	t.Logf("native fill committed: trade rows=1, lastAcceptedHeight=%d", h)
+}
+
+// countTradeRows counts trade:* rows in the VM's durable DB — the authoritative
+// committed fill log. Read under the VM lock (the sealer also touches the DB).
+func countTradeRows(t *testing.T, vm *VM) int {
+	t.Helper()
+	vm.mu.Lock()
+	defer vm.mu.Unlock()
+	it := vm.db.NewIteratorWithPrefix([]byte(prefixTrade))
+	defer it.Release()
+	n := 0
+	for it.Next() {
+		// trade:<height:8><seq:8>
+		if len(it.Key()) == len(prefixTrade)+16 {
+			n++
+		}
+	}
+	if err := it.Error(); err != nil {
+		t.Fatalf("trade iterator: %v", err)
+	}
+	return n
+}
+
+// TestIngestRejectsNonPost proves the boundary contract: the ingestion handler is
+// POST-only (an order is a write); a GET is 405 and never reaches the mempool.
+func TestIngestRejectsNonPost(t *testing.T) {
+	_, base, stop := startIngestServer(t)
+	defer stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, methodURL(base, zapwire.MethodSubmit), nil)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("GET: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusMethodNotAllowed {
+		t.Fatalf("GET status = %d, want 405", resp.StatusCode)
+	}
+}
+
+// TestIngestUnknownMethod404 proves an unrouted method path is a 404 (the mux only
+// mounts the frozen clob_* methods); it must not panic or hang.
+func TestIngestUnknownMethod404(t *testing.T) {
+	_, base, stop := startIngestServer(t)
+	defer stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, methodURL(base, "clob_bogus"), bytes.NewReader([]byte{0x00}))
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("POST: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNotFound {
+		t.Fatalf("unknown method status = %d, want 404", resp.StatusCode)
+	}
+}
+
+// sanity: the ingestion response for a place ack carries the order id in the first
+// 8 bytes big-endian (frozen ack frame), independent of transport.
+func TestIngestAckFrameShape(t *testing.T) {
+	_, base, stop := startIngestServer(t)
+	defer stop()
+
+	var pool [32]byte
+	pool[0] = 0x07
+	postFrame(t, base, zapwire.MethodEnsureMarket, zapwire.EncodeEnsureMarket(pool))
+	ack := postFrame(t, base, zapwire.MethodPlace,
+		zapwire.EncodePlace(pool, zapwire.SideBuy, 5.0, 2.0, "u"))
+	if len(ack) < 9 {
+		t.Fatalf("ack frame too short: %d", len(ack))
+	}
+	if id := binary.BigEndian.Uint64(ack[0:8]); id == 0 {
+		t.Fatal("ack order id is zero")
+	}
+}
@@ -420,9 +420,16 @@ func (vm *VM) Version(ctx context.Context) (string, error) { return Version, nil
 // over the ZAP gateway (handler.go), not HTTP, so this is nil.
 func (vm *VM) NewHTTPHandler(ctx context.Context) (http.Handler, error) { return nil, nil }

-// CreateHandlers returns named HTTP handlers. None — the CLOB surface is ZAP.
+// CreateHandlers returns the VM's named HTTP handlers for luxd to mount under
+// /ext/bc/<DCHAIN_ID>/ (and the "D" alias). It returns one handler per CLOB method
+// keyed by its full sub-path ("/dex/<method>"), so an order POSTed to
+// /ext/bc/D/dex/clob_submit reaches the matcher through the node's own router —
+// the in-luxd ingestion seam (ingest.go). This is how an order enters the native
+// VM: submitTx -> mempool -> consensus -> Verify-match. The plugin transport
+// (github.com/luxfi/vm/rpc) serves these handlers from a local http.Server inside
+// the plugin process and the node reverse-proxies to it.
 func (vm *VM) CreateHandlers(ctx context.Context) (map[string]http.Handler, error) {
-	return map[string]http.Handler{}, nil
+	return vm.httpHandlers(), nil
 }

 // Connected/Disconnected are p2p lifecycle callbacks; the d-chain matcher needs