Skip to content

Commit 0a23c76

Browse files
Exca-DKExca-DK
committed
Implement jsonrpc handler over unix domain socket
1 parent 23bcc52 commit 0a23c76

File tree

9 files changed

+636
-2
lines changed

9 files changed

+636
-2
lines changed

‎cmd/juno/juno.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ const (
4040
wsF = "ws"
4141
wsHostF = "ws-host"
4242
wsPortF = "ws-port"
43+
ipcF = "ipc"
44+
ipcPathF = "ipc-path"
4345
dbPathF = "db-path"
4446
networkF = "network"
4547
ethNodeF = "eth-node"
@@ -69,6 +71,7 @@ const (
6971
defaultHTTPPort = 6060
7072
defaultWS = false
7173
defaultWSPort = 6061
74+
defaultIpc = false
7275
defaultEthNode = ""
7376
defaultPprof = false
7477
defaultPprofPort = 6062
@@ -93,6 +96,8 @@ const (
9396
wsUsage = "Enables the Websocket RPC server on the default port."
9497
wsHostUsage = "The interface on which the Websocket RPC server will listen for requests."
9598
wsPortUsage = "The port on which the websocket server will listen for requests."
99+
ipcUsage = "Enables the IPC RPC server."
100+
ipcPathUsage = "The path on which the IPC RPC server will listen for requests."
96101
dbPathUsage = "Location of the database files."
97102
networkUsage = "Options: mainnet, goerli, goerli2, integration, sepolia, sepolia-integration."
98103
pprofUsage = "Enables the pprof endpoint on the default port."
@@ -219,6 +224,8 @@ func NewCmd(config *node.Config, run func(*cobra.Command, []string) error) *cobr
219224
junoCmd.Flags().Bool(wsF, defaultWS, wsUsage)
220225
junoCmd.Flags().String(wsHostF, defaulHost, wsHostUsage)
221226
junoCmd.Flags().Uint16(wsPortF, defaultWSPort, wsPortUsage)
227+
junoCmd.Flags().Bool(ipcF, defaultIpc, ipcUsage)
228+
junoCmd.Flags().String(ipcPathF, defaultDBPath, ipcPathUsage)
222229
junoCmd.Flags().String(dbPathF, defaultDBPath, dbPathUsage)
223230
junoCmd.Flags().Var(&defaultNetwork, networkF, networkUsage)
224231
junoCmd.Flags().String(ethNodeF, defaultEthNode, ethNodeUsage)

‎cmd/juno/juno_test.go

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ func TestConfigPrecedence(t *testing.T) {
3333
defaultHTTPPort := uint16(6060)
3434
defaultWS := false
3535
defaultWSPort := uint16(6061)
36+
defaultIpc := false
3637
defaultDBPath := filepath.Join(pwd, "juno")
3738
defaultNetwork := utils.Mainnet
3839
defaultPprof := false
@@ -64,6 +65,8 @@ func TestConfigPrecedence(t *testing.T) {
6465
Websocket: defaultWS,
6566
WebsocketHost: defaultHost,
6667
WebsocketPort: defaultWSPort,
68+
IPC: defaultIpc,
69+
IPCPath: defaultDBPath,
6770
DatabasePath: defaultDBPath,
6871
Network: defaultNetwork,
6972
Pprof: defaultPprof,
@@ -93,6 +96,8 @@ func TestConfigPrecedence(t *testing.T) {
9396
Websocket: defaultWS,
9497
WebsocketHost: defaultHost,
9598
WebsocketPort: defaultWSPort,
99+
IPC: defaultIpc,
100+
IPCPath: defaultDBPath,
96101
GRPC: defaultGRPC,
97102
GRPCHost: defaultHost,
98103
GRPCPort: defaultGRPCPort,
@@ -127,6 +132,8 @@ func TestConfigPrecedence(t *testing.T) {
127132
Websocket: defaultWS,
128133
WebsocketHost: defaultHost,
129134
WebsocketPort: defaultWSPort,
135+
IPC: defaultIpc,
136+
IPCPath: defaultDBPath,
130137
GRPC: defaultGRPC,
131138
GRPCHost: defaultHost,
132139
GRPCPort: defaultGRPCPort,
@@ -163,6 +170,8 @@ pprof: true
163170
Websocket: defaultWS,
164171
WebsocketHost: defaultHost,
165172
WebsocketPort: defaultWSPort,
173+
IPC: defaultIpc,
174+
IPCPath: defaultDBPath,
166175
GRPC: defaultGRPC,
167176
GRPCHost: defaultHost,
168177
GRPCPort: defaultGRPCPort,
@@ -196,6 +205,8 @@ http-port: 4576
196205
Websocket: defaultWS,
197206
WebsocketHost: defaultHost,
198207
WebsocketPort: defaultWSPort,
208+
IPC: defaultIpc,
209+
IPCPath: defaultDBPath,
199210
GRPC: defaultGRPC,
200211
GRPCHost: defaultHost,
201212
GRPCPort: defaultGRPCPort,
@@ -228,6 +239,8 @@ http-port: 4576
228239
Websocket: defaultWS,
229240
WebsocketHost: defaultHost,
230241
WebsocketPort: defaultWSPort,
242+
IPC: defaultIpc,
243+
IPCPath: defaultDBPath,
231244
GRPC: defaultGRPC,
232245
GRPCHost: defaultHost,
233246
GRPCPort: defaultGRPCPort,
@@ -259,6 +272,8 @@ http-port: 4576
259272
Websocket: defaultWS,
260273
WebsocketHost: defaultHost,
261274
WebsocketPort: defaultWSPort,
275+
IPC: defaultIpc,
276+
IPCPath: defaultDBPath,
262277
GRPC: defaultGRPC,
263278
GRPCHost: defaultHost,
264279
GRPCPort: defaultGRPCPort,
@@ -287,6 +302,8 @@ http-port: 4576
287302
ws: true
288303
ws-host: 0.0.0.0
289304
ws-port: 4576
305+
ipc: true
306+
ipc-path: /home/config-file/.juno
290307
metrics: true
291308
metrics-host: 0.0.0.0
292309
metrics-port: 4576
@@ -303,7 +320,7 @@ db-cache-size: 8
303320
`,
304321
inputArgs: []string{
305322
"--log-level", "error", "--http", "--http-port", "4577", "--http-host", "127.0.0.1", "--ws", "--ws-port", "4577", "--ws-host", "127.0.0.1",
306-
"--grpc", "--grpc-port", "4577", "--grpc-host", "127.0.0.1", "--metrics", "--metrics-port", "4577", "--metrics-host", "127.0.0.1",
323+
"--ipc", "--ipc-path", "/home/flag/.juno", "--grpc", "--grpc-port", "4577", "--grpc-host", "127.0.0.1", "--metrics", "--metrics-port", "4577", "--metrics-host", "127.0.0.1",
307324
"--db-path", "/home/flag/.juno", "--network", "integration", "--pprof", "--pending-poll-interval", time.Millisecond.String(),
308325
"--db-cache-size", "9",
309326
},
@@ -315,6 +332,8 @@ db-cache-size: 8
315332
Websocket: true,
316333
WebsocketHost: "127.0.0.1",
317334
WebsocketPort: 4577,
335+
IPC: true,
336+
IPCPath: "/home/flag/.juno",
318337
Metrics: true,
319338
MetricsHost: "127.0.0.1",
320339
MetricsPort: 4577,
@@ -350,6 +369,8 @@ network: goerli
350369
Websocket: defaultWS,
351370
WebsocketHost: defaultHost,
352371
WebsocketPort: defaultWSPort,
372+
IPC: defaultIpc,
373+
IPCPath: defaultDBPath,
353374
GRPC: defaultGRPC,
354375
GRPCHost: defaultHost,
355376
GRPCPort: defaultGRPCPort,
@@ -372,7 +393,7 @@ network: goerli
372393
"some setting set in default, config file and flags": {
373394
cfgFile: true,
374395
cfgFileContents: `network: goerli2`,
375-
inputArgs: []string{"--db-path", "/home/flag/.juno", "--pprof"},
396+
inputArgs: []string{"--ipc", "--ipc-path", "/home/flag/.juno", "--db-path", "/home/flag/.juno", "--pprof"},
376397
expectedConfig: &node.Config{
377398
LogLevel: defaultLogLevel,
378399
HTTP: defaultHTTP,
@@ -381,6 +402,8 @@ network: goerli
381402
Websocket: defaultWS,
382403
WebsocketHost: defaultHost,
383404
WebsocketPort: defaultWSPort,
405+
IPC: true,
406+
IPCPath: "/home/flag/.juno",
384407
GRPC: defaultGRPC,
385408
GRPCHost: defaultHost,
386409
GRPCPort: defaultGRPCPort,

‎jsonrpc/ipc.go

Lines changed: 222 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,222 @@
1+
package jsonrpc
2+
3+
import (
4+
"context"
5+
"errors"
6+
"fmt"
7+
"io"
8+
"net"
9+
"os"
10+
"path/filepath"
11+
"sync"
12+
"syscall"
13+
"time"
14+
15+
"github.com/NethermindEth/juno/utils"
16+
"github.com/ethereum/go-ethereum/p2p/netutil"
17+
"github.com/sourcegraph/conc"
18+
)
19+
20+
const (
21+
// On Linux, path is 108 bytes in size.
22+
// http://man7.org/linux/man-pages/man7/unix.7.html
23+
maxUnixPathBytes = 108
24+
// fileModePerms represents the permission mode for directories, which is set to 751.
25+
// This mode is stricter than the common default of 755 for directories.
26+
fileModePerms = 0o751
27+
// socketPerms represents the permission mode for socket files, set to 600. It allows
28+
// read and write access for the owner only, ensuring that socket is accessible only to the intended user.
29+
socketPerms = 0o600
30+
)
31+
32+
// createListener creates a Unix domain socket at the given path and sets proper permissions.
33+
func createListener(endpoint string) (net.Listener, error) {
34+
// path + terminator
35+
if len(endpoint)+1 > maxUnixPathBytes {
36+
return nil, errors.New("path too long")
37+
}
38+
// Try connecting first; if it works, then the socket is still live,
39+
// so let's abort the creation of a new one.
40+
if c, err := net.Dial("unix", endpoint); err == nil {
41+
c.Close()
42+
return nil, fmt.Errorf("%v: address already in use", endpoint)
43+
}
44+
45+
if err := os.MkdirAll(filepath.Dir(endpoint), fileModePerms); err != nil {
46+
return nil, err
47+
}
48+
// Remove any existing file at the specified path.
49+
if err := os.Remove(endpoint); !os.IsNotExist(err) {
50+
return nil, err
51+
}
52+
l, err := net.Listen("unix", endpoint)
53+
if err != nil {
54+
return nil, err
55+
}
56+
// Set permissions for the socket file to read and write for the owner only (0o600)
57+
err = os.Chmod(endpoint, socketPerms)
58+
return l, err
59+
}
60+
61+
type Ipc struct {
62+
rpc *Server
63+
events NewRequestListener
64+
log utils.SimpleLogger
65+
66+
connWg conc.WaitGroup // connWg is a WaitGroup for tracking active connections.
67+
68+
connParams IpcConnParams
69+
listener net.Listener
70+
71+
// everything below is protected
72+
mu sync.Mutex
73+
conns map[net.Conn]struct{} // conns is a map that holds active connections.
74+
}
75+
76+
// NewIpc creates a new IPC handler instance with the provided RPC server, endpoint and logger.
77+
func NewIpc(rpc *Server, endpoint string, log utils.SimpleLogger) (*Ipc, error) {
78+
listener, err := createListener(endpoint)
79+
if err != nil {
80+
return nil, err
81+
}
82+
return &Ipc{
83+
rpc: rpc,
84+
log: log,
85+
connParams: DefaultIpcConnParams(),
86+
conns: make(map[net.Conn]struct{}),
87+
events: &SelectiveListener{},
88+
listener: listener,
89+
}, nil
90+
}
91+
92+
// WithConnParams sanity checks and applies the provided params.
93+
func (i *Ipc) WithConnParams(p *IpcConnParams) *Ipc {
94+
i.connParams = *p
95+
return i
96+
}
97+
98+
// WithListener registers a NewRequestListener
99+
func (i *Ipc) WithListener(listener NewRequestListener) *Ipc {
100+
i.events = listener
101+
return i
102+
}
103+
104+
// Run launches the IPC handler and handles any potential errors.
105+
// It is the caller's responsibility to cancel the provided context when they wish to
106+
// gracefully shut down the IPC handler.
107+
func (i *Ipc) Run(ctx context.Context) error {
108+
var wg conc.WaitGroup
109+
defer wg.Wait()
110+
111+
ctx, cancel := context.WithCancel(ctx)
112+
defer cancel()
113+
114+
wg.Go(func() {
115+
defer cancel()
116+
for {
117+
conn, err := i.listener.Accept()
118+
if netutil.IsTemporaryError(err) {
119+
i.log.Warnw("Failed to accept connection", "err", err)
120+
continue
121+
} else if err != nil {
122+
i.log.Warnw("Accept connection", "err", err)
123+
return
124+
}
125+
i.connWg.Go(func() {
126+
i.serveConn(ctx, newIpcConn(conn, i.connParams))
127+
})
128+
}
129+
})
130+
131+
<-ctx.Done()
132+
return i.stop()
133+
}
134+
135+
// cleanupConnNoLock frees resources
136+
func (i *Ipc) cleanupConnNoLock(conn net.Conn) {
137+
_, ok := i.conns[conn]
138+
delete(i.conns, conn)
139+
if ok {
140+
conn.Close()
141+
}
142+
}
143+
144+
// serveConn handles incoming connection.
145+
func (i *Ipc) serveConn(ctx context.Context, conn net.Conn) {
146+
defer func() {
147+
i.mu.Lock()
148+
defer i.mu.Unlock()
149+
i.cleanupConnNoLock(conn)
150+
}()
151+
i.mu.Lock()
152+
i.conns[conn] = struct{}{}
153+
i.mu.Unlock()
154+
if ctx.Err() != nil {
155+
return
156+
}
157+
158+
var err error
159+
for err == nil {
160+
i.events.OnNewRequest("any")
161+
err = i.rpc.HandleReadWriter(ctx, conn)
162+
}
163+
164+
if isSocketError(err) || isPipeError(err) {
165+
return
166+
}
167+
168+
i.log.Warnw("Closing ipc connection due to internal error", "err", err)
169+
}
170+
171+
// stop gracefully shuts down the IPC handler.
172+
func (i *Ipc) stop() error {
173+
i.mu.Lock()
174+
defer func() {
175+
i.mu.Unlock()
176+
i.connWg.Wait()
177+
}()
178+
err := i.listener.Close()
179+
for conn := range i.conns {
180+
i.cleanupConnNoLock(conn)
181+
}
182+
return err
183+
}
184+
185+
type IpcConnParams struct {
186+
// Maximum time to write a message.
187+
WriteDuration time.Duration
188+
}
189+
190+
type ipcConn struct {
191+
IpcConnParams
192+
net.Conn
193+
}
194+
195+
func DefaultIpcConnParams() IpcConnParams {
196+
return IpcConnParams{
197+
WriteDuration: 5 * time.Second,
198+
}
199+
}
200+
201+
func newIpcConn(conn net.Conn, params IpcConnParams) *ipcConn {
202+
return &ipcConn{
203+
IpcConnParams: params,
204+
Conn: conn,
205+
}
206+
}
207+
208+
func (ipc *ipcConn) Write(p []byte) (int, error) {
209+
if err := ipc.Conn.SetWriteDeadline(time.Now().Add(ipc.WriteDuration)); err != nil {
210+
return 0, err
211+
}
212+
return ipc.Conn.Write(p)
213+
}
214+
215+
func isSocketError(err error) bool {
216+
return errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF)
217+
}
218+
219+
func isPipeError(err error) bool {
220+
// broken pipe || conn reset
221+
return errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET)
222+
}

0 commit comments

Comments
 (0)