maker/coherence_http.go

// Copyright (C) 2019-2026, Lux Industries, Inc. All rights reserved.
// See the file LICENSE for licensing terms.

package maker

import (
	"context"
	"fmt"
	"log/slog"
	"maps"
	"net/http"
	"sort"
	"strings"
	"time"
)

// coherence_http.go exposes the service's measured health as Prometheus text on
// /metrics and a liveness probe on /healthz — the production-ready observability
// surface the k8s Deployment scrapes. Stdlib only; no client_golang dependency for
// a handful of counters and per-symbol gauges.

// baseFeeGweiFloor is the EIP-1559 base-fee floor on the Lux nets (25 gwei,
// measured). Burn = gas · baseFee; on our own L1 the priority tip recycles to the
// validator, so only the base fee is a true burn. Reported in LUX.
const baseFeeGweiFloor = 25.0

// burnLUX converts cumulative gas to the LUX base-fee burn at the 25 gwei floor.
func burnLUX(totalGas uint64) float64 {
	return float64(totalGas) * baseFeeGweiFloor * 1e-9
}

// Prometheus renders the live metrics as Prometheus text exposition format.
func (m *CoherenceMetrics) Prometheus() string {
	m.mu.Lock()
	requotes, arbs, probes, fills, errs, gas := m.Requotes, m.Arbs, m.Probes, m.Fills, m.Errors, m.TotalGas
	reconciles, rebalances := m.Reconciles, m.Rebalances
	uptime := time.Since(m.startedAt).Seconds()
	symbols := make([]string, 0, len(m.lastPegBps))
	for s := range m.lastPegBps {
		symbols = append(symbols, s)
	}
	lastPeg := copyMap(m.lastPegBps)
	lastReal := copyMap(m.lastRealBps)
	venueSpot := copyMap(m.venueSpot)
	venueGap := copyMap(m.venueGapBps)
	m.mu.Unlock()
	sort.Strings(symbols)

	var b strings.Builder
	c := func(name, help string, v uint64) {
		fmt.Fprintf(&b, "# HELP coherence_%s %s\n# TYPE coherence_%s counter\ncoherence_%s %d\n", name, help, name, name, v)
	}
	g := func(name, help string, v float64) {
		fmt.Fprintf(&b, "# HELP coherence_%s %s\n# TYPE coherence_%s gauge\ncoherence_%s %g\n", name, help, name, name, v)
	}
	c("requotes_total", "two-sided book re-pegs to the Kraken mid", requotes)
	c("arbs_total", "corrective arb swaps when the book drifted past the band", arbs)
	c("probes_total", "taker probe trades issued", probes)
	c("fills_total", "settled DEXFills measured", fills)
	c("errors_total", "reference/peg errors (skipped cycles)", errs)
	c("reconciles_total", "cross-venue corrective trades on V3/V4 pools", reconciles)
	c("rebalances_total", "ops-position recenters (incl dry-run plans)", rebalances)
	g("uptime_seconds", "service uptime", uptime)
	g("burn_lux", "cumulative base-fee burn at the 25 gwei floor", burnLUX(gas))
	g("total_gas", "cumulative gas across all service txs", float64(gas))

	// Cross-venue coherence view: last spot + oracle gap per (venue, symbol). The
	// map key is "venue|symbol"; split it into distinct Prometheus labels.
	vkeys := make([]string, 0, len(venueSpot))
	for k := range venueSpot {
		vkeys = append(vkeys, k)
	}
	sort.Strings(vkeys)
	fmt.Fprintf(&b, "# HELP coherence_venue_spot last marginal spot read per venue (quote per base)\n# TYPE coherence_venue_spot gauge\n")
	for _, k := range vkeys {
		venue, symbol := splitVenueKey(k)
		fmt.Fprintf(&b, "coherence_venue_spot{venue=%q,symbol=%q} %g\n", venue, symbol, venueSpot[k])
	}
	fmt.Fprintf(&b, "# HELP coherence_venue_gap_bps last spot-vs-oracle gap per venue (bps)\n# TYPE coherence_venue_gap_bps gauge\n")
	for _, k := range vkeys {
		venue, symbol := splitVenueKey(k)
		fmt.Fprintf(&b, "coherence_venue_gap_bps{venue=%q,symbol=%q} %g\n", venue, symbol, venueGap[k])
	}

	fmt.Fprintf(&b, "# HELP coherence_peg_error_bps last book-mid tracking error vs Kraken (bps)\n# TYPE coherence_peg_error_bps gauge\n")
	for _, s := range symbols {
		fmt.Fprintf(&b, "coherence_peg_error_bps{symbol=%q} %g\n", s, lastPeg[s])
	}
	fmt.Fprintf(&b, "# HELP coherence_peg_error_p50_bps median |book-mid tracking error| over the run (bps)\n# TYPE coherence_peg_error_p50_bps gauge\n")
	for _, s := range symbols {
		fmt.Fprintf(&b, "coherence_peg_error_p50_bps{symbol=%q} %g\n", s, m.pct(s, 0.50))
	}
	fmt.Fprintf(&b, "# HELP coherence_peg_error_p99_bps p99 |book-mid tracking error| over the run (bps)\n# TYPE coherence_peg_error_p99_bps gauge\n")
	for _, s := range symbols {
		fmt.Fprintf(&b, "coherence_peg_error_p99_bps{symbol=%q} %g\n", s, m.pct(s, 0.99))
	}
	fmt.Fprintf(&b, "# HELP coherence_realized_error_bps last realized fill price vs Kraken (bps; ≈ -halfspread for a market order)\n# TYPE coherence_realized_error_bps gauge\n")
	for _, s := range symbols {
		if v, ok := lastReal[s]; ok {
			fmt.Fprintf(&b, "coherence_realized_error_bps{symbol=%q} %g\n", s, v)
		}
	}
	return b.String()
}

// splitVenueKey splits a "venue|symbol" metric key into its two Prometheus labels.
func splitVenueKey(k string) (venue, symbol string) {
	if before, after, ok := strings.Cut(k, "|"); ok {
		return before, after
	}
	return k, ""
}

func copyMap(in map[string]float64) map[string]float64 {
	out := make(map[string]float64, len(in))
	maps.Copy(out, in)
	return out
}

// ServeMetrics starts the /metrics + /healthz HTTP server and blocks until ctx is
// cancelled, then shuts it down gracefully.
func (s *CoherenceService) ServeMetrics(ctx context.Context, addr string, logger *slog.Logger) {
	mux := http.NewServeMux()
	mux.HandleFunc("/metrics", func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "text/plain; version=0.0.4")
		_, _ = w.Write([]byte(s.metrics.Prometheus()))
	})
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write([]byte("ok\n"))
	})
	srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
	go func() {
		<-ctx.Done()
		shutCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutCtx)
	}()
	logger.Info("metrics server listening", "addr", addr)
	if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		logger.Warn("metrics server stopped", "error", err)
	}
}