Fix current node log formats and preserve cursors on discovery errors

CHANGELOG.md

@@ -5,6 +5,15 @@ results in CI or internal records. See docs/releasing.md. -->
 
 ## [Unreleased]
 
+## [v4.1.5] - 2026-10-09
+
+### Fixed
+
+- Read current consensus RPC requests and block responses so schema alerts and served-block metrics work with updated nodes.
+- Avoid replaying older log records when hourly log discovery encounters a directory read error.
+- Accept `Rpc` and `Validator` mempool transaction origins alongside the earlier boolean format.
+- Build with Go 1.26.9 and update `golang.org/x/net` to v0.60.0 for the October security fixes, including the HTTP/2 server crash described in [GO-2026-6617](https://pkg.go.dev/vuln/GO-2026-6617).
+
 ## [v4.1.4] - 2026-10-02
 
 ### Fixed

Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=$BUILDPLATFORM golang:1.25.13-alpine3.23@sha256:42fc3368d1c50170a452f2bf4a1dfd292a065870c3f258d799aad4316671cb69 AS builder
+FROM --platform=$BUILDPLATFORM golang:1.26.9-alpine3.23@sha256:96123126ac58e910f4dd3619a8901e2fb6d1ad84b59b1232cac7c9ea65a8f888 AS builder
 ARG VERSION=docker
 ARG GIT_COMMIT=unknown
 ARG TARGETOS

docs/metrics.md

@@ -83,7 +83,7 @@ Consensus event semantics are source-limited:
 - `RoundCatchUp` direction remains upstream-opaque;
 - accumulator counters add each accepted window's `delta`; `n` is the source-window observation count;
 - registered and sent RPC accumulator values are outbound local work, not served requests;
-- `hl_consensus_rpc_blocks_served_total` counts only the explicitly proven complete outbound-response branch.
+- `hl_consensus_rpc_blocks_served_total` counts blocks explicitly reported by complete inbound RPC lifecycles. Direct requests require the incoming stream, received request and matching outbound response; wrapped requests also require both task stages and `query_peers=false`.
 
 Validator latency EMA is the upstream measured field used by the sampled node build. It is not an exporter-measured protocol RTT. A complete all-zero EMA snapshot is treated as initialization; mixed snapshots preserve genuine zeros.
 
@@ -183,7 +183,7 @@ Run `go generate ./internal/metrics` after changing a declaration. `go test ./in
 | `hl_consensus_round_qc_total` | Counter | Prometheus | - | base | Accepted RoundQc source-window delta accumulated since exporter start; no relationship to committed blocks is inferred. | `prometheus_instruments.go` |
 | `hl_consensus_round_tc_total` | Counter | Prometheus | - | base | Accepted RoundTc source-window delta accumulated since exporter start; no cause or network-health conclusion is inferred. | `prometheus_instruments.go` |
 | `hl_consensus_rounds_per_block` | Gauge | OTel bridge | - | base | Most recent positive difference between consecutive accepted consensus Block round values; a single observed value, not an average | `instruments.go` |
-| `hl_consensus_rpc_blocks_served_total` | Counter | Prometheus | - | base | Blocks explicitly reported sent by complete query_peers=false Outbound response Ok.BlocksAndTxs results since exporter start. | `validator_consensus_prometheus.go` |
+| `hl_consensus_rpc_blocks_served_total` | Counter | Prometheus | - | base | Blocks explicitly reported sent by complete inbound RPC lifecycles since exporter start; wrapped requests require query_peers=false. | `validator_consensus_prometheus.go` |
 | `hl_consensus_rpc_events_total` | Counter | Prometheus | `content`, `direction`, `outcome`, `stage` | base | Consensus RPC lifecycle events since exporter start, classified only into fixed direction, stage, outcome, and content vocabularies. | `validator_consensus_prometheus.go` |
 | `hl_consensus_rpc_parse_total` | Counter | Prometheus | `result` | base | Consensus RPC source records since exporter start by bounded parser result. | `validator_consensus_prometheus.go` |
 | `hl_consensus_rpc_requests_registered_total` | Counter | Prometheus | - | base | Accepted RpcRequestsRegistered delta for outbound local accumulator work since exporter start; not a served-request count. | `prometheus_instruments.go` |

go.mod

@@ -1,6 +1,6 @@
 module github.com/validaoxyz/hyperliquid-exporter
 
-go 1.25.13
+go 1.26.9
 
 require (
 	github.com/joho/godotenv v1.5.1
@@ -31,9 +31,9 @@ require (
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/trace v1.45.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.11.0 // indirect
-	golang.org/x/net v0.58.0 // indirect
-	golang.org/x/sys v0.47.0 // indirect
-	golang.org/x/text v0.41.0 // indirect
+	golang.org/x/net v0.60.0 // indirect
+	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/text v0.42.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect
 	google.golang.org/grpc v1.83.2 // indirect

go.sum

@@ -69,12 +69,12 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
-golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
-golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
-golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8=
-golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M=
+golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU=
+golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
+golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d h1:FarXi840EJWSHYTN3ERkADbPWjl307+FGrA22KAVjjc=

internal/metrics/validator_consensus_prometheus.go

@@ -28,7 +28,7 @@ var (
 	}, []string{"direction", "stage", "outcome", "content"})
 	HLConsensusRPCBlocksServed = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "hl_consensus_rpc_blocks_served_total",
-		Help: "Blocks explicitly reported sent by complete query_peers=false Outbound response Ok.BlocksAndTxs results since exporter start.",
+		Help: "Blocks explicitly reported sent by complete inbound RPC lifecycles since exporter start; wrapped requests require query_peers=false.",
 	})
 	HLConsensusRPCParse = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "hl_consensus_rpc_parse_total",

internal/monitors/consensus_rpc_monitor.go

@@ -1,10 +1,12 @@
 package monitors
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -24,6 +26,7 @@ type consensusRPCRecord struct {
 	outcome     string
 	blocks      int64
 	queryPeers  bool
+	direct      bool
 	hasSession  bool
 	hasRequest  bool
 	hasResponse bool
@@ -39,6 +42,7 @@ type consensusRPCLifecycle struct {
 	request      [32]byte
 	content      string
 	received     bool
+	direct       bool
 	taskInbound  bool
 	taskResponse bool
 	response     [32]byte
@@ -150,11 +154,14 @@ func parseConsensusRPCLine(line []byte) (consensusRPCRecord, error) {
 		if len(payload) <= index {
 			return fmt.Errorf("missing request")
 		}
-		digest, content, queryPeers, err := parseConsensusRPCRequest(payload[index])
+		digest, content, queryPeers, direct, err := parseConsensusRPCRequest(payload[index])
 		if err != nil {
 			return err
 		}
-		record.request, record.content, record.queryPeers = digest, content, queryPeers
+		if direct && tag != "Received rpc request" && tag != "Outbound response" {
+			return fmt.Errorf("direct request at unsupported stage")
+		}
+		record.request, record.content, record.queryPeers, record.direct = digest, content, queryPeers, direct
 		record.hasRequest = true
 		return nil
 	}
@@ -236,33 +243,98 @@ func parseConsensusRPCLine(line []byte) (consensusRPCRecord, error) {
 	return record, nil
 }
 
-func parseConsensusRPCRequest(raw json.RawMessage) ([32]byte, string, bool, error) {
-	var body map[string]json.RawMessage
-	if err := json.Unmarshal(raw, &body); err != nil || body == nil {
-		return [32]byte{}, "", false, fmt.Errorf("invalid request object")
-	}
-	var queryPeers bool
-	queryRaw, ok := body["query_peers"]
-	if !ok || unmarshalRequiredJSON(queryRaw, &queryPeers) != nil {
-		return [32]byte{}, "", false, fmt.Errorf("invalid query_peers")
-	}
-	contentRaw, ok := body["content"]
-	if !ok {
-		return [32]byte{}, "", false, fmt.Errorf("missing request content")
+func parseConsensusRPCRequest(raw json.RawMessage) ([32]byte, string, bool, bool, error) {
+	body, err := consensusRPCObject(raw)
+	if err != nil {
+		return [32]byte{}, "", false, false, fmt.Errorf("invalid request object: %w", err)
 	}
+	direct := false
+	queryPeers := false
 	content := "other"
-	var contentObject map[string]json.RawMessage
-	if err := json.Unmarshal(contentRaw, &contentObject); err != nil || contentObject == nil {
-		return [32]byte{}, "", false, fmt.Errorf("invalid request content")
-	}
-	if _, ok := contentObject["BlocksAndTxs"]; ok {
-		content = "blocks_and_txs"
+	if _, wrapped := body["content"]; wrapped {
+		if _, mixed := body["after_round"]; mixed {
+			return [32]byte{}, "", false, false, fmt.Errorf("conflicting request forms")
+		}
+		if _, mixed := body["until_block_hash"]; mixed {
+			return [32]byte{}, "", false, false, fmt.Errorf("conflicting request forms")
+		}
+		if unmarshalRequiredJSON(body["query_peers"], &queryPeers) != nil {
+			return [32]byte{}, "", false, false, fmt.Errorf("invalid query_peers")
+		}
+		contentObject, err := consensusRPCObject(body["content"])
+		if err != nil {
+			return [32]byte{}, "", false, false, fmt.Errorf("invalid request content")
+		}
+		_, hasRound := contentObject["after_round"]
+		_, hasHash := contentObject["until_block_hash"]
+		if hasRound || hasHash {
+			if !validFlatConsensusRPCRequest(contentObject) {
+				return [32]byte{}, "", false, false, fmt.Errorf("invalid flat request content")
+			}
+			content = "blocks_and_txs"
+		} else {
+			if len(contentObject) != 1 {
+				return [32]byte{}, "", false, false, fmt.Errorf("invalid request variants")
+			}
+			if blocksRaw, ok := contentObject["BlocksAndTxs"]; ok {
+				blocksBody, err := consensusRPCObject(blocksRaw)
+				if err != nil || !validFlatConsensusRPCRequest(blocksBody) {
+					return [32]byte{}, "", false, false, fmt.Errorf("invalid BlocksAndTxs request")
+				}
+				content = "blocks_and_txs"
+			}
+		}
+	} else {
+		if !validFlatConsensusRPCRequest(body) {
+			return [32]byte{}, "", false, false, fmt.Errorf("invalid direct request")
+		}
+		direct, content = true, "blocks_and_txs"
 	}
 	canonical, err := canonicalJSON(raw)
 	if err != nil {
-		return [32]byte{}, "", false, err
+		return [32]byte{}, "", false, false, err
+	}
+	return sha256.Sum256(canonical), content, queryPeers, direct, nil
+}
+
+func validFlatConsensusRPCRequest(body map[string]json.RawMessage) bool {
+	var round uint64
+	var hash string
+	return len(body) == 2 && unmarshalRequiredJSON(body["after_round"], &round) == nil &&
+		unmarshalRequiredJSON(body["until_block_hash"], &hash) == nil && hash != ""
+}
+
+func consensusRPCObject(raw json.RawMessage) (map[string]json.RawMessage, error) {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	if token, err := decoder.Token(); err != nil || token != json.Delim('{') {
+		return nil, fmt.Errorf("expected object")
+	}
+	body := make(map[string]json.RawMessage)
+	for decoder.More() {
+		token, err := decoder.Token()
+		if err != nil {
+			return nil, err
+		}
+		key, ok := token.(string)
+		if !ok {
+			return nil, fmt.Errorf("invalid object key")
+		}
+		if _, exists := body[key]; exists {
+			return nil, fmt.Errorf("duplicate object key")
+		}
+		var value json.RawMessage
+		if err := decoder.Decode(&value); err != nil {
+			return nil, err
+		}
+		body[key] = value
+	}
+	if token, err := decoder.Token(); err != nil || token != json.Delim('}') {
+		return nil, fmt.Errorf("unclosed object")
+	}
+	if _, err := decoder.Token(); err != io.EOF {
+		return nil, fmt.Errorf("trailing object data")
 	}
-	return sha256.Sum256(canonical), content, queryPeers, nil
+	return body, nil
 }
 
 func parseConsensusRPCResponse(raw json.RawMessage) ([32]byte, string, string, int64, error) {
@@ -271,25 +343,44 @@ func parseConsensusRPCResponse(raw json.RawMessage) ([32]byte, string, string, i
 		return [32]byte{}, "", "", 0, err
 	}
 	digest := sha256.Sum256(canonical)
-	var variants map[string]json.RawMessage
-	if err := json.Unmarshal(raw, &variants); err != nil || variants == nil {
+	variants, err := consensusRPCObject(raw)
+	if err != nil {
 		return [32]byte{}, "", "", 0, fmt.Errorf("invalid response object")
 	}
+	if len(variants) != 1 {
+		return [32]byte{}, "", "", 0, fmt.Errorf("invalid response variants")
+	}
 	okRaw, ok := variants["Ok"]
 	if !ok {
 		return digest, "other", "other", 0, nil
 	}
-	var okBody map[string]json.RawMessage
-	if err := json.Unmarshal(okRaw, &okBody); err != nil || okBody == nil {
+	okBody, err := consensusRPCObject(okRaw)
+	if err != nil {
 		return [32]byte{}, "", "", 0, fmt.Errorf("invalid Ok response")
 	}
 	blocksRaw, ok := okBody["BlocksAndTxs"]
-	if !ok {
-		return digest, "other", "other", 0, nil
-	}
+	_, hasN := okBody["n"]
+	_, hasHash := okBody["last_block_hash"]
 	var blocksBody map[string]json.RawMessage
-	if err := json.Unmarshal(blocksRaw, &blocksBody); err != nil || blocksBody == nil {
-		return [32]byte{}, "", "", 0, fmt.Errorf("invalid BlocksAndTxs response")
+	if ok {
+		if len(okBody) != 1 {
+			return [32]byte{}, "", "", 0, fmt.Errorf("conflicting block response forms")
+		}
+		blocksBody, err = consensusRPCObject(blocksRaw)
+		if err != nil {
+			return [32]byte{}, "", "", 0, fmt.Errorf("invalid BlocksAndTxs response")
+		}
+	} else if hasN || hasHash {
+		var hash string
+		if len(okBody) != 2 || unmarshalRequiredJSON(okBody["last_block_hash"], &hash) != nil || hash == "" {
+			return [32]byte{}, "", "", 0, fmt.Errorf("invalid flat block response")
+		}
+		blocksBody = okBody
+	} else {
+		if len(okBody) != 1 {
+			return [32]byte{}, "", "", 0, fmt.Errorf("invalid Ok response variants")
+		}
+		return digest, "other", "other", 0, nil
 	}
 	var n int64
 	if rawN, present := blocksBody["n"]; !present || unmarshalRequiredJSON(rawN, &n) != nil || n < 1 || n > maxConsensusRPCServedBlocks {
@@ -300,7 +391,9 @@ func parseConsensusRPCResponse(raw json.RawMessage) ([32]byte, string, string, i
 
 func canonicalJSON(raw []byte) ([]byte, error) {
 	var value interface{}
-	if err := json.Unmarshal(raw, &value); err != nil {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.UseNumber()
+	if err := decoder.Decode(&value); err != nil {
 		return nil, err
 	}
 	return json.Marshal(value)
@@ -323,11 +416,14 @@ func (t *consensusRPCTracker) accept(record consensusRPCRecord) string {
 		if life == nil || life.received || record.queryPeers {
 			return "unjoinable"
 		}
-		if existing, collision := t.byRequest[record.request]; collision && existing != record.session {
+		if existing, collision := t.byRequest[record.request]; !record.direct && collision && existing != record.session {
 			return "unjoinable"
 		}
 		life.request, life.content, life.received, life.updatedAt = record.request, record.content, true, now
-		t.byRequest[record.request] = record.session
+		life.direct = record.direct
+		if !record.direct {
+			t.byRequest[record.request] = record.session
+		}
 		return "ok"
 	case "Rpc task inbound":
 		life := t.lifecycleForRequest(record.request)
@@ -345,8 +441,8 @@ func (t *consensusRPCTracker) accept(record consensusRPCRecord) string {
 		return "ok"
 	case "Outbound response":
 		life := t.bySession[record.session]
-		if life == nil || !life.received || !life.taskInbound || !life.taskResponse ||
-			life.request != record.request || life.response != record.response || record.queryPeers {
+		if life == nil || !life.received || life.direct != record.direct || life.request != record.request || record.queryPeers ||
+			(!life.direct && (!life.taskInbound || !life.taskResponse || life.response != record.response)) {
 			return "unjoinable"
 		}
 		completionKey := sha256.Sum256(append(record.session[:], record.response[:]...))
@@ -364,17 +460,21 @@ func (t *consensusRPCTracker) accept(record consensusRPCRecord) string {
 			delete(t.completed, oldestKey)
 		}
 		t.completed[completionKey] = now
-		// Emit stage counters only after the complete five-record lifecycle
-		// joins. An abandoned or replayed prefix must not look like served work.
+		// Publish only after the full observed lifecycle joins. Direct requests
+		// have no task stages; wrapped requests still require both task records.
 		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "stream_opened", "observed", "other").Inc()
 		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "request_received", "observed", life.content).Inc()
-		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_inbound", "observed", life.content).Inc()
-		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_response", "observed", life.content).Inc()
+		if !life.direct {
+			metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_inbound", "observed", life.content).Inc()
+			metrics.HLConsensusRPCEvents.WithLabelValues("serve", "task_response", "observed", life.content).Inc()
+		}
 		metrics.HLConsensusRPCEvents.WithLabelValues("serve", "response_sent", record.outcome, record.content).Inc()
 		if record.outcome == "ok" && record.content == "blocks_and_txs" {
 			metrics.HLConsensusRPCBlocksServed.Add(float64(record.blocks))
 		}
-		delete(t.byRequest, life.request)
+		if !life.direct {
+			delete(t.byRequest, life.request)
+		}
 		delete(t.bySession, life.session)
 		return "ok"
 	default:
@@ -395,7 +495,9 @@ func (t *consensusRPCTracker) expire(now time.Time) {
 	cutoff := now.Add(-t.ttl)
 	for session, life := range t.bySession {
 		if life.updatedAt.Before(cutoff) {
-			delete(t.byRequest, life.request)
+			if !life.direct {
+				delete(t.byRequest, life.request)
+			}
 			delete(t.bySession, session)
 		}
 	}