fix: log warning when solana-exec collector hits sig limit

harnesses/aggregator-head-lag/cmd/script/defined_auth.go

@@ -58,6 +58,52 @@ func decodeJWTExpiration(token string) (time.Time, error) {
 	return time.Unix(claims.Exp, 0), nil
 }
 
+// tryDirectCodexToken calls /api/codex/token on defined.fi directly using the session cookie.
+// No proxy so the JWE is minted from this container's own IP, matching the WS connection IP.
+func tryDirectCodexToken(sessionCookie string) (string, error) {
+	if sessionCookie == "" {
+		return "", fmt.Errorf("no session cookie")
+	}
+	client := &http.Client{
+		Timeout: 10 * time.Second,
+		Transport: &http.Transport{
+			DisableKeepAlives: true,
+		},
+	}
+	req, err := http.NewRequest("GET", "https://www.defined.fi/api/codex/token", nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("Accept-Language", "en-US,en;q=0.9")
+	req.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36")
+	req.Header.Set("Origin", "https://www.defined.fi")
+	req.Header.Set("Referer", "https://www.defined.fi/")
+	req.Header.Set("sec-ch-ua", `"Not_A Brand";v="8", "Chromium";v="131", "Google Chrome";v="131"`)
+	req.Header.Set("sec-ch-ua-mobile", "?0")
+	req.Header.Set("sec-ch-ua-platform", `"macOS"`)
+	req.Header.Set("sec-fetch-dest", "empty")
+	req.Header.Set("sec-fetch-mode", "cors")
+	req.Header.Set("sec-fetch-site", "same-origin")
+	req.AddCookie(&http.Cookie{Name: "defined-attestation-token", Value: sessionCookie})
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("request failed: %w", err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != 200 {
+		return "", fmt.Errorf("status %d: %.100s", resp.StatusCode, string(body))
+	}
+	var parsed struct {
+		Token string `json:"token"`
+	}
+	if err := json.Unmarshal(body, &parsed); err == nil && parsed.Token != "" {
+		return parsed.Token, nil
+	}
+	return "", fmt.Errorf("no token in response: %.100s", string(body))
+}
+
 // tryTokenService calls the Paris-box sidecar (DEFINED_TOKEN_SERVICE_URL) for a fresh JWE.
 func tryTokenService(baseURL string) (string, error) {
 	client := &http.Client{Timeout: 5 * time.Second}
@@ -84,12 +130,17 @@ func tryTokenService(baseURL string) (string, error) {
 }
 
 // GetDefinedJWTToken returns a cached JWT token or generates a new one if expired.
-// Priority: CODEX_JWT env var > DEFINED_TOKEN_SERVICE_URL sidecar > inline mint.
+// Priority: CODEX_JWT env var > direct /api/codex/token (same IP as WS) > sidecar > inline mint.
 func GetDefinedJWTToken(sessionCookie string) (string, error) {
 	if jwt := os.Getenv("CODEX_JWT"); jwt != "" {
 		return jwt, nil
 	}
-	// Sidecar token service (Paris box chromedp scraper, auto-refreshes every 25 min)
+	// Direct mint: JWE minted from this container's IP = same IP used for WS = no 4403.
+	if tok, err := tryDirectCodexToken(sessionCookie); err == nil && tok != "" {
+		fmt.Printf("[DEFINED-AUTH] Got token via direct /api/codex/token (len=%d)\n", len(tok))
+		return tok, nil
+	}
+	// Sidecar fallback (Paris box chromedp, auto-refreshes every 25 min)
 	if svcURL := os.Getenv("DEFINED_TOKEN_SERVICE_URL"); svcURL != "" {
 		if tok, err := tryTokenService(svcURL); err == nil && tok != "" {
 			fmt.Printf("[DEFINED-AUTH] Got token from sidecar (len=%d)\n", len(tok))

harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"net/http"
+	"net/url"
 	"strings"
 	"sync"
 	"time"
@@ -419,11 +421,19 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error {
 		return fmt.Errorf("failed to get JWT token: %w", err)
 	}
 
-	log.Printf("[HEAD-LAG][CODEX] Step 2/4: Creating proxy dialer...")
-	dialer := getProxyDialerWithSubprotocols([]string{"graphql-transport-ws"})
+	log.Printf("[HEAD-LAG][CODEX] Step 2/4: Creating direct dialer (no proxy — IP must match JWE origin)...")
+	dialer := &websocket.Dialer{
+		Subprotocols:     []string{"graphql-transport-ws"},
+		HandshakeTimeout: 30 * time.Second,
+	}
 
 	log.Printf("[HEAD-LAG][CODEX] Step 3/4: Connecting to wss://graph.codex.io/graphql...")
-	conn, resp, err := dialer.Dial("wss://graph.codex.io/graphql", nil)
+	cookieVal := url.QueryEscape(`{"token":"` + jwtToken + `"}`)
+	wsHeaders := http.Header{}
+	wsHeaders.Set("Cookie", "codex_token="+cookieVal)
+	wsHeaders.Set("Origin", "https://www.defined.fi")
+	wsHeaders.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36")
+	conn, resp, err := dialer.Dial("wss://graph.codex.io/graphql", wsHeaders)
 	if err != nil {
 		if resp != nil {
 			return fmt.Errorf("dial failed (HTTP %d): %w", resp.StatusCode, err)
@@ -439,12 +449,11 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error {
 		log.Printf("[HEAD-LAG][CODEX] Step 3/4: ✅ WebSocket connection established (IP check failed: %v)", err)
 	}
 
-	// Connection init with JWT Bearer token
-	log.Printf("[HEAD-LAG][CODEX] Step 4/4: Sending connection_init with JWT...")
+	log.Printf("[HEAD-LAG][CODEX] Step 4/4: Sending connection_init (token len=%d)...", len(jwtToken))
 	initMsg := map[string]interface{}{
 		"type": "connection_init",
 		"payload": map[string]interface{}{
-			"Authorization": fmt.Sprintf("Bearer %s", jwtToken),
+			"Authorization": "Bearer " + jwtToken,
 		},
 	}
 	if err := conn.WriteJSON(initMsg); err != nil {

harnesses/solana-exec/cmd/api/main.go

@@ -64,29 +64,42 @@ func handleExecLeaderboard(pool *pgxpool.Pool) http.HandlerFunc {
 		rows, err := pool.Query(ctx, `
 			SELECT
 				platform,
-				AVG(avg_priority_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_prio,
-				AVG(p50_cu_price_micro)         FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p50,
-				AVG(p95_cu_price_micro)         FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p95,
-				AVG(avg_platform_fee_lamports)  FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_pfee,
-				AVG(jito_rate)                  FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_jito,
-				AVG(avg_cu_consumed)            FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_cu,
-				SUM(tx_count)                   FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_count,
-
-				AVG(avg_priority_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_prio,
-				AVG(p50_cu_price_micro)         FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_p50,
-				AVG(p95_cu_price_micro)         FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_p95,
-				AVG(avg_platform_fee_lamports)  FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_pfee,
-				AVG(jito_rate)                  FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_jito,
-				AVG(avg_cu_consumed)            FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_cu,
-				SUM(tx_count)                   FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')   AS d7_count,
-
-				AVG(avg_priority_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_prio,
-				AVG(p50_cu_price_micro)         FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_p50,
-				AVG(p95_cu_price_micro)         FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_p95,
-				AVG(avg_platform_fee_lamports)  FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_pfee,
-				AVG(jito_rate)                  FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_jito,
-				AVG(avg_cu_consumed)            FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_cu,
-				SUM(tx_count)                   FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')  AS d30_count,
+				-- weighted averages: SUM(avg×count)/SUM(count) avoids skewing by small off-peak buckets
+				SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_prio,
+				AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p50,
+				AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p95,
+				SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_pfee,
+				SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_jito,
+				SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_cu,
+				SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_count,
+
+				SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_prio,
+				AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p50,
+				AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p95,
+				SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_pfee,
+				SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_jito,
+				SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_cu,
+				SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_count,
+
+				SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_prio,
+				AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p50,
+				AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p95,
+				SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_pfee,
+				SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_jito,
+				SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_cu,
+				SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_count,
 
 				MAX(bucket_start)::text AS latest_bucket
 			FROM solana_exec_facts

harnesses/solana-exec/cmd/collector/main.go

@@ -44,14 +44,22 @@ func collect(ctx context.Context, db *store.DB, h *helius.Client, plt, feeAccoun
 		return fmt.Errorf("get cursor: %w", err)
 	}
 
+	const sigLimit = 100
 	// Fetch signatures newer than last seen. Results are newest-first.
-	sigs, err := h.GetSignaturesForAddress(ctx, feeAccount, 100, cursor.LastSig)
+	sigs, err := h.GetSignaturesForAddress(ctx, feeAccount, sigLimit, cursor.LastSig)
 	if err != nil {
 		return fmt.Errorf("get sigs: %w", err)
 	}
 	if len(sigs) == 0 {
 		return nil
 	}
+	// On incremental polls (cursor set), hitting the cap means we're dropping older
+	// txs from this window — tx counts will be understated, though fee stats remain
+	// a representative sample of the most-recent transactions.
+	// The initial bootstrap (no cursor) always hits the cap; that's expected.
+	if len(sigs) == sigLimit && cursor.LastSig != "" {
+		log.Printf("collector: %s: WARNING hit sig limit (%d) — older txs in this poll window dropped; reduce POLL_INTERVAL or increase limit", plt, sigLimit)
+	}
 
 	// Reverse to process oldest-first so cursor is always the true watermark.
 	reversed := make([]helius.SigEntry, len(sigs))