|
1 | 1 | package node |
2 | 2 |
|
3 | 3 | import ( |
4 | | - "context" |
5 | | - "encoding/json" |
6 | | - "fmt" |
7 | | - "net/url" |
8 | | - "time" |
| 4 | + "context" |
| 5 | + "encoding/json" |
| 6 | + "fmt" |
| 7 | + "net/url" |
| 8 | + "time" |
9 | 9 |
|
10 | | - "github.com/gorilla/websocket" |
| 10 | + "github.com/gorilla/websocket" |
11 | 11 | ) |
12 | 12 |
|
13 | 13 | // DialAndSubscribeHeaders uses gorilla/websocket to subscribe to NewBlockHeader events and stream heights. |
14 | 14 | func DialAndSubscribeHeaders(ctx context.Context, wsURL string) (<-chan Header, error) { |
15 | | - u, err := url.Parse(wsURL) |
16 | | - if err != nil { return nil, err } |
17 | | - if u.Path == "" { u.Path = "/websocket" } |
| 15 | + u, err := url.Parse(wsURL) |
| 16 | + if err != nil { |
| 17 | + return nil, err |
| 18 | + } |
| 19 | + if u.Path == "" { |
| 20 | + u.Path = "/websocket" |
| 21 | + } |
18 | 22 |
|
19 | | - d := websocket.Dialer{ |
20 | | - Subprotocols: []string{"jsonrpc"}, |
21 | | - HandshakeTimeout: 5 * time.Second, |
22 | | - EnableCompression: false, |
23 | | - } |
24 | | - // nolint:bodyclose |
25 | | - conn, _, err := d.DialContext(ctx, u.String(), map[string][]string{"Origin": {"http://localhost"}}) |
26 | | - if err != nil { return nil, err } |
| 23 | + d := websocket.Dialer{ |
| 24 | + Subprotocols: []string{"jsonrpc"}, |
| 25 | + HandshakeTimeout: 5 * time.Second, |
| 26 | + EnableCompression: false, |
| 27 | + } |
| 28 | + // nolint:bodyclose |
| 29 | + conn, _, err := d.DialContext(ctx, u.String(), map[string][]string{"Origin": {"http://localhost"}}) |
| 30 | + if err != nil { |
| 31 | + return nil, err |
| 32 | + } |
27 | 33 |
|
28 | | - // Send subscribe request |
29 | | - sub := map[string]any{ |
30 | | - "jsonrpc": "2.0", |
31 | | - "method": "subscribe", |
32 | | - // Prefer cometbft.event key for 0.38+; servers typically support both. |
33 | | - "params": map[string]string{"query": "cometbft.event='NewBlockHeader'"}, |
34 | | - "id": 1, |
35 | | - } |
36 | | - if err := conn.WriteJSON(sub); err != nil { _ = conn.Close(); return nil, err } |
| 34 | + // Send subscribe request |
| 35 | + sub := map[string]any{ |
| 36 | + "jsonrpc": "2.0", |
| 37 | + "method": "subscribe", |
| 38 | + // Prefer cometbft.event key for 0.38+; servers typically support both. |
| 39 | + "params": map[string]string{"query": "cometbft.event='NewBlockHeader'"}, |
| 40 | + "id": 1, |
| 41 | + } |
| 42 | + if err := conn.WriteJSON(sub); err != nil { |
| 43 | + _ = conn.Close() |
| 44 | + return nil, err |
| 45 | + } |
37 | 46 |
|
38 | | - out := make(chan Header, 32) |
39 | | - go func() { |
40 | | - defer close(out) |
41 | | - defer func() { |
42 | | - // attempt proper close handshake |
43 | | - deadline := time.Now().Add(1500 * time.Millisecond) |
44 | | - _ = conn.SetWriteDeadline(deadline) |
45 | | - _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), deadline) |
46 | | - // best-effort wait for server close |
47 | | - _ = conn.SetReadDeadline(deadline) |
48 | | - _, _, _ = conn.ReadMessage() |
49 | | - _ = conn.Close() |
50 | | - }() |
51 | | - for { |
52 | | - select { |
53 | | - case <-ctx.Done(): |
54 | | - return |
55 | | - default: |
56 | | - } |
57 | | - // Read next message |
58 | | - // Read next message (gorilla handles ping/pong) |
59 | | - _, msg, err := conn.ReadMessage() |
60 | | - if err != nil { |
61 | | - // graceful exits on normal closure or going away |
62 | | - if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { |
63 | | - return |
64 | | - } |
65 | | - // any other error: return and let defer handle close |
66 | | - return |
67 | | - } |
68 | | - if h, ok := parseHeaderHeight(msg); ok { out <- h; continue } |
69 | | - // handle pong/ping implicitly via gorilla; ignore others |
70 | | - } |
71 | | - }() |
72 | | - return out, nil |
| 47 | + // Read deadline is a safety net for TCP connections that don't close properly. |
| 48 | + // During block sync, WS silence for minutes is normal (events only at consensus tip). |
| 49 | + const readTimeout = 5 * time.Minute |
| 50 | + |
| 51 | + out := make(chan Header, 32) |
| 52 | + go func() { |
| 53 | + defer close(out) |
| 54 | + defer func() { |
| 55 | + // attempt proper close handshake |
| 56 | + deadline := time.Now().Add(1500 * time.Millisecond) |
| 57 | + _ = conn.SetWriteDeadline(deadline) |
| 58 | + _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), deadline) |
| 59 | + // best-effort wait for server close |
| 60 | + _ = conn.SetReadDeadline(deadline) |
| 61 | + _, _, _ = conn.ReadMessage() |
| 62 | + _ = conn.Close() |
| 63 | + }() |
| 64 | + for { |
| 65 | + select { |
| 66 | + case <-ctx.Done(): |
| 67 | + return |
| 68 | + default: |
| 69 | + } |
| 70 | + // Set read deadline before each read; reset on success |
| 71 | + _ = conn.SetReadDeadline(time.Now().Add(readTimeout)) |
| 72 | + _, msg, err := conn.ReadMessage() |
| 73 | + if err != nil { |
| 74 | + // graceful exits on normal closure or going away |
| 75 | + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { |
| 76 | + return |
| 77 | + } |
| 78 | + // any other error (timeout, connection reset): exit |
| 79 | + return |
| 80 | + } |
| 81 | + if h, ok := parseHeaderHeight(msg); ok { |
| 82 | + out <- h |
| 83 | + continue |
| 84 | + } |
| 85 | + // handle pong/ping implicitly via gorilla; ignore others |
| 86 | + } |
| 87 | + }() |
| 88 | + return out, nil |
73 | 89 | } |
74 | 90 |
|
75 | 91 | func parseHeaderHeight(b []byte) (Header, bool) { |
76 | | - var payload struct { |
77 | | - Result struct { |
78 | | - Data struct { |
79 | | - Value struct { |
80 | | - Header struct { |
81 | | - Height string `json:"height"` |
82 | | - Time time.Time `json:"time"` |
83 | | - } `json:"header"` |
84 | | - } `json:"value"` |
85 | | - } `json:"data"` |
86 | | - } `json:"result"` |
87 | | - } |
88 | | - if err := json.Unmarshal(b, &payload); err != nil { return Header{}, false } |
89 | | - if payload.Result.Data.Value.Header.Height == "" { return Header{}, false } |
90 | | - // Accept both tm.event and cometbft.event streams; height parse only |
91 | | - h, err := strconvParseInt(payload.Result.Data.Value.Header.Height) |
92 | | - if err != nil { return Header{}, false } |
93 | | - return Header{Height: h, Time: payload.Result.Data.Value.Header.Time}, true |
| 92 | + var payload struct { |
| 93 | + Result struct { |
| 94 | + Data struct { |
| 95 | + Value struct { |
| 96 | + Header struct { |
| 97 | + Height string `json:"height"` |
| 98 | + Time time.Time `json:"time"` |
| 99 | + } `json:"header"` |
| 100 | + } `json:"value"` |
| 101 | + } `json:"data"` |
| 102 | + } `json:"result"` |
| 103 | + } |
| 104 | + if err := json.Unmarshal(b, &payload); err != nil { |
| 105 | + return Header{}, false |
| 106 | + } |
| 107 | + if payload.Result.Data.Value.Header.Height == "" { |
| 108 | + return Header{}, false |
| 109 | + } |
| 110 | + // Accept both tm.event and cometbft.event streams; height parse only |
| 111 | + h, err := strconvParseInt(payload.Result.Data.Value.Header.Height) |
| 112 | + if err != nil { |
| 113 | + return Header{}, false |
| 114 | + } |
| 115 | + return Header{Height: h, Time: payload.Result.Data.Value.Header.Time}, true |
94 | 116 | } |
95 | 117 |
|
96 | 118 | func strconvParseInt(s string) (int64, error) { |
97 | | - var n int64 |
98 | | - var sign int64 = 1 |
99 | | - if s == "" { return 0, fmt.Errorf("empty") } |
100 | | - if s[0] == '-' { sign = -1; s = s[1:] } |
101 | | - for i := 0; i < len(s); i++ { |
102 | | - c := s[i] |
103 | | - if c < '0' || c > '9' { return 0, fmt.Errorf("invalid") } |
104 | | - n = n*10 + int64(c-'0') |
105 | | - } |
106 | | - return sign * n, nil |
| 119 | + var n int64 |
| 120 | + var sign int64 = 1 |
| 121 | + if s == "" { |
| 122 | + return 0, fmt.Errorf("empty") |
| 123 | + } |
| 124 | + if s[0] == '-' { |
| 125 | + sign = -1 |
| 126 | + s = s[1:] |
| 127 | + } |
| 128 | + for i := 0; i < len(s); i++ { |
| 129 | + c := s[i] |
| 130 | + if c < '0' || c > '9' { |
| 131 | + return 0, fmt.Errorf("invalid") |
| 132 | + } |
| 133 | + n = n*10 + int64(c-'0') |
| 134 | + } |
| 135 | + return sign * n, nil |
107 | 136 | } |
0 commit comments