Implement max number of entries per bson (#56)

* implement max number of entries per bson

* add env

* fix env name

* fix typo

.env.example

@@ -3,6 +3,7 @@ MONGO_INITDB_ROOT_PASSWORD=
 MONGO_DB_API_PORT=
 API_PORT=
 LOG_LEVEL=
+MAX_ENTRIES_PER_BSON=
 BEACON_NODE_URL_MAINNET=
 BEACON_NODE_URL_HOLESKY=
 BEACON_NODE_URL_GNOSIS=

README.md

@@ -7,11 +7,12 @@ This repository contains the code for the validator monitoring system. The syste
 In dappnode, the signature request and validation flow is as follows:
 
 1. The staking brain sends the signature request of type `PROOF_OF_VALIDATION` to the web3signer (see https://github.com/Consensys/web3signer/pull/982). The request has the following format:
+
 ```json
 {
-  "type" : "PROOF_OF_VALIDATION",
-  "platform" : "dappnode",
-  "timestamp" : "1711338489397"
+  "type": "PROOF_OF_VALIDATION",
+  "platform": "dappnode",
+  "timestamp": "1711338489397"
 }
 ```
 
@@ -22,15 +23,16 @@ In dappnode, the signature request and validation flow is as follows:
 ## API
 
 - `/signatures?network=<network>`:
-    - `POST`: TODO
-    - `GET`: TODO
+  - `POST`: TODO
+  - `GET`: TODO
 
 ## Validation
 
 The process of validating the request and the signature follows the next steps:
 
-1. Get network query parameter from the request: it is mandatory and must be one of "mainnet", "holesy", "gnosis", "lukso". 
+1. Get network query parameter from the request: it is mandatory and must be one of "mainnet", "holesy", "gnosis", "lukso".
 2. Decode and validate the request. The request body must have the following format:
+
 ```go
 type SignatureRequest struct {
 	Payload   string `json:"payload"`
@@ -39,17 +41,20 @@ type SignatureRequest struct {
 	Tag       Tag    `json:"tag"`
 }
 ```
+
 The payload must be encoded in base64 and must have the following format:
+
 ```go
 type DecodedPayload struct {
 	Type      string `json:"type"`
 	Platform  string `json:"platform"`
 	Timestamp string `json:"timestamp"`
 }
 ```
-3. The validators must be in status "active_on_going" according to a standard beacon node API, see https://ethereum.github.io/beacon-APIs/#/Beacon/postStateValidators: 
-    3.1 The signatures from the validators that are not in this status will be discarded. 
-    3.2 If in the moment of querying the beacon node to get the validator status the beacon node is down the signature will be accepted storing the validator status as "unknown" for later validation.
+
+3. The validators must be in status "active_on_going" according to a standard beacon node API, see https://ethereum.github.io/beacon-APIs/#/Beacon/postStateValidators:
+   3.1 The signatures from the validators that are not in this status will be discarded.
+   3.2 If in the moment of querying the beacon node to get the validator status the beacon node is down the signature will be accepted storing the validator status as "unknown" for later validation.
 4. Only the signatures that have passed the previous steps will be validated. The validation of the signature will be done using the pubkey from the request.
 5. Only valid signatures will be stored in the database.
 
@@ -59,15 +64,16 @@ There are 2 cron to ensure the system is working properly:
 
 - `removeOldSignatures`: this cron will remove from the db signatures older than 30 days
 - `updateSignaturesStatus`:
-    - This cron will update the status of the validators that are in status "unknown" to "active_on_going" if the validator is active in the beacon node. 
-    - If the beacon node is down the status will remain as "unknown". 
-    - If the validator is not active the signature will be removed from the database.
+  - This cron will update the status of the validators that are in status "unknown" to "active_on_going" if the validator is active in the beacon node.
+  - If the beacon node is down the status will remain as "unknown".
+  - If the validator is not active the signature will be removed from the database.
 
 ## Database
 
 The database is a mongo db that stores the signatures as BSON's. There are considered as unique the combination of the following fields: `network`, `pubkey`, `tag`. In order to keep the size of the database as small as possible there is a `entries` collection that stores the payload signature and decodedPayload of each request.
 
 The BSON of each unique validator has the following format:
+
 ```go
 bson.M{
     "pubkey":  req.Pubkey,
@@ -99,6 +105,7 @@ MONGO_INITDB_ROOT_PASSWORD=
 MONGO_DB_API_PORT=
 API_PORT=
 LOG_LEVEL=
+MAX_ENTRIES_PER_BSON= # It is recommended to set a low value like 100 for this variable since mongo db has a limit of 16MB per document
 BEACON_NODE_URL_MAINNET=
 BEACON_NODE_URL_HOLESKY=
 BEACON_NODE_URL_GNOSIS=
@@ -115,7 +122,6 @@ docker compose -f docker-compose.dev.yml up -d --scale brain=5
 
 The flag `--scale brain=5` is optional and it will run 5 instances of the staking brain in order to simulate a real environment.
 
-
 ## Running the system
 
 **Requirements:**

docker-compose.dev.yml

@@ -83,9 +83,14 @@ services:
           - postgres.web3signer.dappnode
 
   brain:
-    build:
-      context: web3signer/brain
-      dockerfile: Dockerfile
+    image: staking-brain:0.1.18
+    environment:
+      NETWORK: holesky
+      _DAPPNODE_GLOBAL_EXECUTION_CLIENT_HOLESKY: "holesky-geth.dnp.dappnode.eth"
+      _DAPPNODE_GLOBAL_CONSENSUS_CLIENT_HOLESKY: "lighthouse-holesky.dnp.dappnode.eth"
+      SHARE_DATA_WITH_DAPPNODE: "true"
+      VALIDATORS_MONITOR_URL: "http://listener.dappnode:8080"
+      SHARE_CRON_INTERVAL: "3600"
     restart: always
     volumes:
       - "brain_data:/app/data"

docker-compose.yml

@@ -13,6 +13,7 @@ services:
       BEACON_NODE_URL_HOLESKY: ${BEACON_NODE_URL_HOLESKY}
       BEACON_NODE_URL_LUKSO: ${BEACON_NODE_URL_LUKSO}
       BEACON_NODE_URL_GNOSIS: ${BEACON_NODE_URL_GNOSIS}
+      MAX_ENTRIES_PER_BSON: ${MAX_ENTRIES_PER_BSON}
     depends_on:
       - mongo
     container_name: listener

listener/cmd/listener/main.go

@@ -21,7 +21,7 @@ import (
 func main() {
 	logger.Info("Starting listener")
 	// Load config
-	config, err := config.LoadConfig()
+	config, err := config.GetConfig()
 	if err != nil {
 		logger.Fatal("Failed to load config: " + err.Error())
 	}
@@ -51,6 +51,7 @@ func main() {
 		dbClient,
 		dbCollection,
 		config.BeaconNodeURLs,
+		config.MaxEntriesPerBson,
 	)
 
 	// Start the API server in a goroutine. Needs to be in a goroutine to allow for the cron job to run,

listener/internal/api/api.go

@@ -12,20 +12,22 @@ import (
 )
 
 type httpApi struct {
-	server         *http.Server
-	port           string
-	dbClient       *mongo.Client
-	dbCollection   *mongo.Collection
-	beaconNodeUrls map[types.Network]string
+	server            *http.Server
+	port              string
+	dbClient          *mongo.Client
+	dbCollection      *mongo.Collection
+	beaconNodeUrls    map[types.Network]string
+	maxEntriesPerBson int
 }
 
 // create a new api instance
-func NewApi(port string, dbClient *mongo.Client, dbCollection *mongo.Collection, beaconNodeUrls map[types.Network]string) *httpApi {
+func NewApi(port string, dbClient *mongo.Client, dbCollection *mongo.Collection, beaconNodeUrls map[types.Network]string, maxEntriesPerBson int) *httpApi {
 	return &httpApi{
-		port:           port,
-		dbClient:       dbClient,
-		dbCollection:   dbCollection,
-		beaconNodeUrls: beaconNodeUrls,
+		port:              port,
+		dbClient:          dbClient,
+		dbCollection:      dbCollection,
+		beaconNodeUrls:    beaconNodeUrls,
+		maxEntriesPerBson: maxEntriesPerBson,
 	}
 }
 
@@ -39,7 +41,7 @@ func (s *httpApi) Start() {
 
 	s.server = &http.Server{
 		Addr:    ":" + s.port,
-		Handler: routes.SetupRouter(s.dbCollection, s.beaconNodeUrls),
+		Handler: routes.SetupRouter(s.dbCollection, s.beaconNodeUrls, s.maxEntriesPerBson),
 	}
 
 	// ListenAndServe returns ErrServerClosed to indicate that the server has been shut down when the server is closed gracefully. We need to

listener/internal/api/handlers/postSignatures.go

@@ -3,6 +3,8 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"net/http"
 
 	"github.com/dappnode/validator-monitoring/listener/internal/api/types"
@@ -13,7 +15,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
-func PostSignatures(w http.ResponseWriter, r *http.Request, dbCollection *mongo.Collection, beaconNodeUrls map[types.Network]string) {
+func PostSignatures(w http.ResponseWriter, r *http.Request, dbCollection *mongo.Collection, beaconNodeUrls map[types.Network]string, maxEntriesPerBson int) {
 	logger.Debug("Received new POST '/signatures' request")
 	var requests []types.SignatureRequest
 
@@ -61,9 +63,9 @@ func PostSignatures(w http.ResponseWriter, r *http.Request, dbCollection *mongo.
 	}
 
 	// Insert valid signatures into MongoDB
-	if err := insertSignaturesIntoDB(validSignatures, network, dbCollection); err != nil {
+	if err := insertSignaturesIntoDB(validSignatures, network, dbCollection, maxEntriesPerBson); err != nil {
 		logger.Error("Failed to insert signatures into MongoDB: " + err.Error())
-		respondError(w, http.StatusInternalServerError, "Failed to insert signatures into MongoDB")
+		respondError(w, http.StatusInternalServerError, "Failed to insert signatures into MongoDB: "+err.Error())
 		return
 	}
 
@@ -103,14 +105,29 @@ func filterAndVerifySignatures(requests []types.SignatureRequestDecoded, validat
 	return validSignatures
 }
 
-func insertSignaturesIntoDB(signatures []types.SignatureRequestDecodedWithStatus, network types.Network, dbCollection *mongo.Collection) error {
+func insertSignaturesIntoDB(signatures []types.SignatureRequestDecodedWithStatus, network types.Network, dbCollection *mongo.Collection, maxEntriesPerBson int) error {
 	for _, req := range signatures {
 		filter := bson.M{
 			"pubkey":  req.Pubkey,
 			"tag":     req.Tag,
 			"network": network,
 		}
 
+		// Check the number of entries
+		var result struct {
+			Entries []bson.M `bson:"entries"`
+		}
+		err := dbCollection.FindOne(context.Background(), filter).Decode(&result)
+		if err != nil && err != mongo.ErrNoDocuments {
+			return err
+		}
+
+		// mongo DB has a limit of 16MB per document
+		// if this limit is reached the following exception is thrown: `write exception: write errors: [Resulting document after update is larger than 16777216]`
+		if len(result.Entries) >= maxEntriesPerBson {
+			return errors.New("Max number of entries reached for pubkey " + req.Pubkey + ". Max entries per pubkey: " + fmt.Sprint(maxEntriesPerBson))
+		}
+
 		// Create a base update document with $push operation
 		update := bson.M{
 			"$push": bson.M{

listener/internal/api/routes/routes.go

@@ -9,14 +9,14 @@ import (
 	"go.mongodb.org/mongo-driver/mongo"
 )
 
-func SetupRouter(dbCollection *mongo.Collection, beaconNodeUrls map[types.Network]string) *mux.Router {
+func SetupRouter(dbCollection *mongo.Collection, beaconNodeUrls map[types.Network]string, maxEntriesPerBson int) *mux.Router {
 	r := mux.NewRouter()
 
 	// Define routes
 	r.HandleFunc("/", handlers.GetHealthCheck).Methods(http.MethodGet)
 	// closure function to inject dbCollection into the handler
 	r.HandleFunc("/signatures", func(w http.ResponseWriter, r *http.Request) {
-		handlers.PostSignatures(w, r, dbCollection, beaconNodeUrls)
+		handlers.PostSignatures(w, r, dbCollection, beaconNodeUrls, maxEntriesPerBson)
 	}).Methods(http.MethodPost)
 
 	// Middlewares

listener/internal/config/config.go → listener/internal/config/getConfig.go

@@ -3,6 +3,7 @@ package config
 import (
 	"fmt"
 	"os"
+	"strconv"
 
 	"github.com/dappnode/validator-monitoring/listener/internal/api/types"
 	"github.com/dappnode/validator-monitoring/listener/internal/logger"
@@ -18,9 +19,11 @@ type Config struct {
 	LogLevel string
 	// BeaconNodeURLs is the URLs of the beacon nodes for different networks
 	BeaconNodeURLs map[types.Network]string
+	// Max number of entries allowed per BSON document
+	MaxEntriesPerBson int
 }
 
-func LoadConfig() (*Config, error) {
+func GetConfig() (*Config, error) {
 	logLevel := os.Getenv("LOG_LEVEL")
 	if logLevel == "" {
 		logger.Info("LOG_LEVEL is not set, using default INFO")
@@ -55,9 +58,25 @@ func LoadConfig() (*Config, error) {
 	if beaconLukso == "" {
 		return nil, fmt.Errorf("BEACON_NODE_URL_LUKSO is not set")
 	}
+	maxEntriesPerBsonStr := os.Getenv("MAX_ENTRIES_PER_BSON")
+	if maxEntriesPerBsonStr == "" {
+		logger.Info("MAX_ENTRIES_PER_BSON is not set, using default 30")
+		maxEntriesPerBsonStr = "30"
+	}
+	MaxEntriesPerBson, err := strconv.Atoi(maxEntriesPerBsonStr)
+	if err != nil {
+		return nil, fmt.Errorf("MAX_ENTRIES_PER_BSON is not a valid integer")
+	}
 
-	// print all envs in a single line
-	logger.Info("Loaded config: LOG_LEVEL=" + logLevel + " API_PORT=" + apiPort + " MONGO_DB_URI=" + mongoDBURI + " BEACON_NODE_URL_MAINNET=" + beaconMainnet + " BEACON_NODE_URL_HOLESKY=" + beaconHolesky + " BEACON_NODE_URL_GNOSIS=" + beaconGnosis + " BEACON_NODE_URL_LUKSO=" + beaconLukso)
+	// print all envs beauty with newlines
+	logger.Info("LOG_LEVEL: " + logLevel)
+	logger.Info("API_PORT: " + apiPort)
+	logger.Info("MONGO_DB_URI: " + mongoDBURI)
+	logger.Info("BEACON_NODE_URL_MAINNET: " + beaconMainnet)
+	logger.Info("BEACON_NODE_URL_HOLESKY: " + beaconHolesky)
+	logger.Info("BEACON_NODE_URL_GNOSIS: " + beaconGnosis)
+	logger.Info("BEACON_NODE_URL_LUKSO: " + beaconLukso)
+	logger.Info("MAX_ENTRIES_PER_BSON: " + maxEntriesPerBsonStr)
 
 	beaconNodeURLs := map[types.Network]string{
 		types.Mainnet: beaconMainnet,
@@ -67,9 +86,10 @@ func LoadConfig() (*Config, error) {
 	}
 
 	return &Config{
-		Port:           apiPort,
-		MongoDBURI:     mongoDBURI,
-		LogLevel:       logLevel,
-		BeaconNodeURLs: beaconNodeURLs,
+		Port:              apiPort,
+		MongoDBURI:        mongoDBURI,
+		LogLevel:          logLevel,
+		BeaconNodeURLs:    beaconNodeURLs,
+		MaxEntriesPerBson: MaxEntriesPerBson,
 	}, nil
 }