fix: max allowance check session validate worker

Author: ironman0x7b2

cmd/root.go

@@ -73,7 +73,7 @@ node initialization, ensuring privacy, performance, and ease of use.`,
 	)
 
 	// Persistent flags
-	rootCmd.PersistentFlags().String("home", homeDir, "Home directory for application config and data")
+	rootCmd.PersistentFlags().String("home", homeDir, "home directory for application config and data")
 
 	return rootCmd
 }

config/node.go

@@ -240,14 +240,14 @@ func DefaultNodeConfig() NodeConfig {
 		APIPort:                                fmt.Sprintf("%d", utils.RandomPort()),
 		GigabytePrices:                         "0.01;0;udvpn",
 		HourlyPrices:                           "0.02;0;udvpn",
-		IntervalBestRPCAddr:                    "5m",
-		IntervalGeoIPLocation:                  "6h",
-		IntervalSessionUsageSyncWithBlockchain: "2h",
-		IntervalSessionUsageSyncWithDatabase:   "5s",
-		IntervalSessionUsageValidate:           "5s",
-		IntervalSessionValidate:                "1m",
-		IntervalSpeedtest:                      "168h",
-		IntervalStatusUpdate:                   "1h",
+		IntervalBestRPCAddr:                    (5 * time.Minute).String(),
+		IntervalGeoIPLocation:                  (6 * time.Hour).String(),
+		IntervalSessionUsageSyncWithBlockchain: (2*time.Hour - 5*time.Minute).String(),
+		IntervalSessionUsageSyncWithDatabase:   (3 * time.Second).String(),
+		IntervalSessionUsageValidate:           (3 * time.Second).String(),
+		IntervalSessionValidate:                (5 * time.Minute).String(),
+		IntervalSpeedtest:                      (7 * 24 * time.Hour).String(),
+		IntervalStatusUpdate:                   (1*time.Hour - 5*time.Minute).String(),
 		Moniker:                                randMoniker(),
 		RemoteAddrs:                            []string{},
 		Type:                                   randServiceType().String(),

node/node.go

@@ -1,7 +1,7 @@
 package node
 
 import (
-	"context"
+	gocontext "context"
 	"errors"
 	"fmt"
 	"path/filepath"
@@ -11,7 +11,7 @@ import (
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
 	"github.com/sentinel-official/sentinel-go-sdk/libs/log"
 
-	nodecontext "github.com/sentinel-official/dvpn-node/context"
+	"github.com/sentinel-official/dvpn-node/context"
 	"github.com/sentinel-official/dvpn-node/utils"
 )
 
@@ -61,8 +61,8 @@ func (n *Node) Scheduler() *cron.Scheduler {
 }
 
 // Register registers the node on the network if not already registered.
-func (n *Node) Register(c *nodecontext.Context) error {
-	node, err := c.Client().Node(context.TODO(), c.AccAddr().Bytes())
+func (n *Node) Register(c *context.Context) error {
+	node, err := c.Client().Node(gocontext.TODO(), c.NodeAddr())
 	if err != nil {
 		return fmt.Errorf("failed to query node: %w", err)
 	}
@@ -74,14 +74,14 @@ func (n *Node) Register(c *nodecontext.Context) error {
 
 	// Prepare a message to register the node.
 	msg := v3.NewMsgRegisterNodeRequest(
-		c.AccAddr().Bytes(),
+		c.AccAddr(),
 		c.GigabytePrices(),
 		c.HourlyPrices(),
 		c.RemoteAddrs()[0],
 	)
 
 	// Broadcast the registration transaction.
-	res, err := c.BroadcastTx(context.TODO(), msg)
+	res, err := c.BroadcastTx(gocontext.TODO(), msg)
 	if err != nil {
 		return fmt.Errorf("failed to broadcast register node tx: %w", err)
 	}
@@ -94,19 +94,19 @@ func (n *Node) Register(c *nodecontext.Context) error {
 }
 
 // UpdateDetails updates the node's pricing and address details on the network.
-func (n *Node) UpdateDetails(c *nodecontext.Context) error {
+func (n *Node) UpdateDetails(c *context.Context) error {
 	log.Info("Updating node details...")
 
 	// Prepare a message to update the node's details.
 	msg := v3.NewMsgUpdateNodeDetailsRequest(
-		c.AccAddr().Bytes(),
+		c.NodeAddr(),
 		c.GigabytePrices(),
 		c.HourlyPrices(),
 		c.RemoteAddrs()[0],
 	)
 
 	// Broadcast the update transaction.
-	res, err := c.BroadcastTx(context.TODO(), msg)
+	res, err := c.BroadcastTx(gocontext.TODO(), msg)
 	if err != nil {
 		return fmt.Errorf("failed to broadcast update node details tx: %w", err)
 	}
@@ -119,12 +119,12 @@ func (n *Node) UpdateDetails(c *nodecontext.Context) error {
 }
 
 // Start initializes the Node's services, scheduler, and HTTPS server.
-func (n *Node) Start(c *nodecontext.Context, errChan chan error) error {
+func (n *Node) Start(c *context.Context, errChan chan error) error {
 	log.Info("Starting node...")
 
 	go func() {
 		// Bring up the service by running pre-defined tasks.
-		if err := c.Service().Up(context.TODO()); err != nil {
+		if err := c.Service().Up(gocontext.TODO()); err != nil {
 			errChan <- fmt.Errorf("failed to run service up task: %w", err)
 			return
 		}
@@ -166,11 +166,11 @@ func (n *Node) Start(c *nodecontext.Context, errChan chan error) error {
 }
 
 // Stop gracefully stops the Node's operations.
-func (n *Node) Stop(c *nodecontext.Context) error {
+func (n *Node) Stop(c *context.Context) error {
 	if err := c.Service().PreDown(); err != nil {
 		return fmt.Errorf("failed to run service pre-down task: %w", err)
 	}
-	if err := c.Service().Down(context.TODO()); err != nil {
+	if err := c.Service().Down(gocontext.TODO()); err != nil {
 		return fmt.Errorf("failed to run service down task: %w", err)
 	}
 	if err := c.Service().PostDown(); err != nil {

workers/node.go

@@ -1,7 +1,8 @@
 package workers
 
 import (
-	"context"
+	gocontext "context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -10,14 +11,14 @@ import (
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
 	logger "github.com/sentinel-official/sentinel-go-sdk/libs/log"
 
-	nodecontext "github.com/sentinel-official/dvpn-node/context"
+	"github.com/sentinel-official/dvpn-node/context"
 )
 
 const nameNodeStatusUpdate = "node_status_update"
 
 // NewNodeStatusUpdateWorker creates a worker to periodically update the node's status to active on the blockchain.
 // This worker broadcasts a transaction to mark the node as active at regular intervals.
-func NewNodeStatusUpdateWorker(c *nodecontext.Context, interval time.Duration) cron.Worker {
+func NewNodeStatusUpdateWorker(c *context.Context, interval time.Duration) cron.Worker {
 	log := logger.With("name", nameNodeStatusUpdate)
 
 	// Handler function that updates the node's status to active.
@@ -31,12 +32,13 @@ func NewNodeStatusUpdateWorker(c *nodecontext.Context, interval time.Duration) c
 		)
 
 		// Broadcast the transaction message to the blockchain.
-		res, err := c.BroadcastTx(context.TODO(), msg)
+		res, err := c.BroadcastTx(gocontext.TODO(), msg)
 		if err != nil {
 			return fmt.Errorf("failed to broadcast update node status tx: %w", err)
 		}
 		if !res.TxResult.IsOK() {
-			return fmt.Errorf("update node status tx failed with code %d: %s", res.TxResult.Code, res.TxResult.Log)
+			err := errors.New(res.TxResult.Log)
+			return fmt.Errorf("update node status tx failed with code %d: %w", res.TxResult.Code, err)
 		}
 
 		return nil

workers/session.go

@@ -1,7 +1,8 @@
 package workers
 
 import (
-	"context"
+	gocontext "context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -11,7 +12,7 @@ import (
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
 	logger "github.com/sentinel-official/sentinel-go-sdk/libs/log"
 
-	nodecontext "github.com/sentinel-official/dvpn-node/context"
+	"github.com/sentinel-official/dvpn-node/context"
 	"github.com/sentinel-official/dvpn-node/database/operations"
 )
 
@@ -25,7 +26,7 @@ const (
 // NewSessionUsageSyncWithBlockchainWorker creates a worker that synchronizes session usage with the blockchain.
 // This worker retrieves session data from the database, validates it against the blockchain,
 // and broadcasts any updates as transactions.
-func NewSessionUsageSyncWithBlockchainWorker(c *nodecontext.Context, interval time.Duration) cron.Worker {
+func NewSessionUsageSyncWithBlockchainWorker(c *context.Context, interval time.Duration) cron.Worker {
 	log := logger.With("name", nameSessionUsageSyncWithBlockchain)
 
 	handlerFunc := func() error {
@@ -34,7 +35,7 @@ func NewSessionUsageSyncWithBlockchainWorker(c *nodecontext.Context, interval ti
 		// Retrieve session records from the database.
 		items, err := operations.SessionFind(c.Database(), nil)
 		if err != nil {
-			return fmt.Errorf("failed to retrieve sessions from database: %w", err)
+			return fmt.Errorf("failed to retrieve sessions from the database: %w", err)
 		}
 		if len(items) == 0 {
 			return nil
@@ -43,9 +44,9 @@ func NewSessionUsageSyncWithBlockchainWorker(c *nodecontext.Context, interval ti
 		var msgs []types.Msg
 		// Iterate over sessions and prepare messages for updates.
 		for _, item := range items {
-			session, err := c.Client().Session(context.TODO(), item.GetID())
+			session, err := c.Client().Session(gocontext.TODO(), item.GetID())
 			if err != nil {
-				return fmt.Errorf("failed to query session: %w", err)
+				return fmt.Errorf("failed to query session from the blockchain: %w", err)
 			}
 
 			if session != nil {
@@ -56,12 +57,13 @@ func NewSessionUsageSyncWithBlockchainWorker(c *nodecontext.Context, interval ti
 		}
 
 		// Broadcast the prepared messages as a transaction.
-		res, err := c.BroadcastTx(context.TODO(), msgs...)
+		res, err := c.BroadcastTx(gocontext.TODO(), msgs...)
 		if err != nil {
-			return fmt.Errorf("failed to broadcast update session details tx: %w", err)
+			return fmt.Errorf("failed to broadcast update session tx: %w", err)
 		}
 		if !res.TxResult.IsOK() {
-			return fmt.Errorf("update session details tx failed with code %d: %s", res.TxResult.Code, res.TxResult.Log)
+			err := errors.New(res.TxResult.Log)
+			return fmt.Errorf("update session tx failed with code %d: %w", res.TxResult.Code, err)
 		}
 
 		return nil
@@ -83,13 +85,14 @@ func NewSessionUsageSyncWithBlockchainWorker(c *nodecontext.Context, interval ti
 
 // NewSessionUsageSyncWithDatabaseWorker creates a worker that updates session usage in the database.
 // This worker fetches usage data from the peer service and updates the corresponding database records.
-func NewSessionUsageSyncWithDatabaseWorker(c *nodecontext.Context, interval time.Duration) cron.Worker {
+func NewSessionUsageSyncWithDatabaseWorker(c *context.Context, interval time.Duration) cron.Worker {
 	log := logger.With("name", nameSessionUsageSyncWithDatabase)
 
 	handlerFunc := func() error {
 		log.Info("Running scheduler worker")
 
-		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		// Create a context with a timeout to fetch peer statistics.
+		ctx, cancel := gocontext.WithTimeout(gocontext.TODO(), 5*time.Second)
 		defer cancel()
 
 		// Fetch peer usage statistics from the service.
@@ -103,19 +106,24 @@ func NewSessionUsageSyncWithDatabaseWorker(c *nodecontext.Context, interval time
 
 		// Update the database with the fetched statistics.
 		for _, item := range items {
+			// Convert usage statistics to strings for database storage.
 			downloadBytes := math.NewInt(item.DownloadBytes).String()
 			uploadBytes := math.NewInt(item.UploadBytes).String()
 
+			// Define query to find the session by peer key.
 			query := map[string]interface{}{
 				"peer_key": item.Key,
 			}
+
+			// Define updates to apply to the session record.
 			updates := map[string]interface{}{
 				"download_bytes": downloadBytes,
 				"upload_bytes":   uploadBytes,
 			}
 
+			// Update the session in the database.
 			if _, err := operations.SessionFindOneAndUpdate(c.Database(), query, updates); err != nil {
-				return fmt.Errorf("failed to update session: %w", err)
+				return fmt.Errorf("failed to update session with peer key %s: %w", item.Key, err)
 			}
 		}
 
@@ -138,7 +146,7 @@ func NewSessionUsageSyncWithDatabaseWorker(c *nodecontext.Context, interval time
 
 // NewSessionUsageValidateWorker creates a worker that validates session usage limits and removes peers if necessary.
 // This worker checks if sessions exceed their maximum byte or duration limits and removes peers accordingly.
-func NewSessionUsageValidateWorker(c *nodecontext.Context, interval time.Duration) cron.Worker {
+func NewSessionUsageValidateWorker(c *context.Context, interval time.Duration) cron.Worker {
 	log := logger.With("name", nameSessionUsageValidate)
 
 	handlerFunc := func() error {
@@ -147,7 +155,7 @@ func NewSessionUsageValidateWorker(c *nodecontext.Context, interval time.Duratio
 		// Retrieve session records from the database.
 		items, err := operations.SessionFind(c.Database(), nil)
 		if err != nil {
-			return fmt.Errorf("failed to retrieve sessions from database: %w", err)
+			return fmt.Errorf("failed to retrieve sessions from the database: %w", err)
 		}
 		if len(items) == 0 {
 			return nil
@@ -157,16 +165,22 @@ func NewSessionUsageValidateWorker(c *nodecontext.Context, interval time.Duratio
 		for _, item := range items {
 			removePeer := false
 
-			if item.GetBytes().GTE(item.GetMaxBytes()) {
+			// Check if the session exceeds the maximum allowed bytes.
+			maxBytes := item.GetMaxBytes()
+			if !maxBytes.IsZero() && item.GetBytes().GTE(maxBytes) {
 				removePeer = true
 			}
-			if item.GetDuration() >= item.GetMaxDuration() {
+
+			// Check if the session exceeds the maximum allowed duration.
+			maxDuration := item.GetMaxDuration()
+			if maxDuration != 0 && item.GetDuration() >= maxDuration {
 				removePeer = true
 			}
 
+			// Remove the peer if validation fails.
 			if removePeer {
-				if err := c.RemovePeerIfExistsForKey(context.TODO(), item.PeerKey); err != nil {
-					return fmt.Errorf("failed to remove peer: %w", err)
+				if err := c.RemovePeerIfExistsForKey(gocontext.TODO(), item.PeerKey); err != nil {
+					return fmt.Errorf("failed to remove peer with key %s: %w", item.PeerKey, err)
 				}
 			}
 		}
@@ -190,7 +204,7 @@ func NewSessionUsageValidateWorker(c *nodecontext.Context, interval time.Duratio
 
 // NewSessionValidateWorker creates a worker that validates session status and removes peers if necessary.
 // This worker ensures sessions are active and consistent between the database and blockchain.
-func NewSessionValidateWorker(c *nodecontext.Context, interval time.Duration) cron.Worker {
+func NewSessionValidateWorker(c *context.Context, interval time.Duration) cron.Worker {
 	log := logger.With("name", nameSessionValidate)
 
 	handlerFunc := func() error {
@@ -199,23 +213,23 @@ func NewSessionValidateWorker(c *nodecontext.Context, interval time.Duration) cr
 		// Retrieve session records from the database.
 		items, err := operations.SessionFind(c.Database(), nil)
 		if err != nil {
-			return fmt.Errorf("failed to retrieve sessions from database: %w", err)
+			return fmt.Errorf("failed to retrieve sessions from the database: %w", err)
 		}
 		if len(items) == 0 {
 			return nil
 		}
 
 		// Validate session status and consistency.
 		for _, item := range items {
-			session, err := c.Client().Session(context.TODO(), item.GetID())
+			session, err := c.Client().Session(gocontext.TODO(), item.GetID())
 			if err != nil {
-				return fmt.Errorf("failed to query session: %w", err)
+				return fmt.Errorf("failed to query session from the blockchain: %w", err)
 			}
 
 			// Remove peers if sessions are inactive or missing on the blockchain.
 			if session == nil || !session.GetStatus().Equal(v1.StatusActive) {
-				if err := c.RemovePeerIfExistsForKey(context.TODO(), item.PeerKey); err != nil {
-					return fmt.Errorf("failed to remove peer: %w", err)
+				if err := c.RemovePeerIfExistsForKey(gocontext.TODO(), item.PeerKey); err != nil {
+					return fmt.Errorf("failed to remove peer with key %s: %w", item.PeerKey, err)
 				}
 			}
 
@@ -224,8 +238,9 @@ func NewSessionValidateWorker(c *nodecontext.Context, interval time.Duration) cr
 				query := map[string]interface{}{
 					"id": item.ID,
 				}
+
 				if _, err := operations.SessionFindOneAndDelete(c.Database(), query); err != nil {
-					return fmt.Errorf("failed to delete session: %w", err)
+					return fmt.Errorf("failed to delete session %d: %w", item.ID, err)
 				}
 			}
 		}