Increase FNS default numbers and close connections properly with reasons

dydxprotocol · Dec 9, 2024 · 0ef04b2 · 0ef04b2
1 parent fc5ff36
commit 0ef04b2
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 17 deletions.
diff --git a/protocol/app/flags/flags.go b/protocol/app/flags/flags.go
@@ -72,8 +72,8 @@ const (
 
 	DefaultGrpcStreamingEnabled              = false
 	DefaultGrpcStreamingFlushIntervalMs      = 50
-	DefaultGrpcStreamingMaxBatchSize         = 10000
-	DefaultGrpcStreamingMaxChannelBufferSize = 10000
+	DefaultGrpcStreamingMaxBatchSize         = 100_000
+	DefaultGrpcStreamingMaxChannelBufferSize = 100_000
 	DefaultWebsocketStreamingEnabled         = false
 	DefaultWebsocketStreamingPort            = 9092
 	DefaultFullNodeStreamingSnapshotInterval = 0

diff --git a/protocol/streaming/ws/websocket_server.go b/protocol/streaming/ws/websocket_server.go
@@ -19,6 +19,8 @@ import (
 const (
 	CLOB_PAIR_IDS_QUERY_PARAM = "clobPairIds"
 	MARKET_IDS_QUERY_PARAM    = "marketIds"
+
+	CLOSE_DEADLINE = 5 * time.Second
 )
 
 var upgrader = websocket.Upgrader{
@@ -68,33 +70,24 @@ func (ws *WebsocketServer) Handler(w http.ResponseWriter, r *http.Request) {
 	// Parse clobPairIds from query parameters
 	clobPairIds, err := parseUint32(r, CLOB_PAIR_IDS_QUERY_PARAM)
 	if err != nil {
-		ws.logger.Error(
-			"Error parsing clobPairIds",
-			"err", err,
-		)
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		ws.logger.Error("Error parsing clobPairIds", "err", err)
+		sendCloseWithReason(conn, websocket.CloseUnsupportedData, err.Error())
 		return
 	}
 
 	// Parse marketIds from query parameters
 	marketIds, err := parseUint32(r, MARKET_IDS_QUERY_PARAM)
 	if err != nil {
-		ws.logger.Error(
-			"Error parsing marketIds",
-			"err", err,
-		)
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		ws.logger.Error("Error parsing marketIds", "err", err)
+		sendCloseWithReason(conn, websocket.CloseUnsupportedData, err.Error())
 		return
 	}
 
 	// Parse subaccountIds from query parameters
 	subaccountIds, err := parseSubaccountIds(r)
 	if err != nil {
-		ws.logger.Error(
-			"Error parsing subaccountIds",
-			"err", err,
-		)
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		ws.logger.Error("Error parsing subaccountIds", "err", err)
+		sendCloseWithReason(conn, websocket.CloseUnsupportedData, err.Error())
 		return
 	}
 
@@ -118,10 +111,22 @@ func (ws *WebsocketServer) Handler(w http.ResponseWriter, r *http.Request) {
 			"Ending handler for websocket connection",
 			"err", err,
 		)
+		sendCloseWithReason(conn, websocket.CloseInternalServerErr, err.Error())
 		return
 	}
 }
 
+func sendCloseWithReason(conn *websocket.Conn, closeCode int, reason string) error {
+	closeMessage := websocket.FormatCloseMessage(closeCode, reason)
+	// Set a write deadline to avoid blocking indefinitely
+	conn.SetWriteDeadline(time.Now().Add(CLOSE_DEADLINE))
+	return conn.WriteControl(
+		websocket.CloseMessage,
+		closeMessage,
+		time.Now().Add(CLOSE_DEADLINE),
+	)
+}
+
 // parseSubaccountIds is a helper function to parse the subaccountIds from the query parameters.
 func parseSubaccountIds(r *http.Request) ([]*satypes.SubaccountId, error) {
 	subaccountIdsParam := r.URL.Query().Get("subaccountIds")