chore: added debug and info logs in workers

Author: ironman0x7b2

core/service.go

@@ -23,6 +23,6 @@ func (c *Context) RemovePeerIfExists(ctx context.Context, id string) error {
 		return fmt.Errorf("removing peer %q from service: %w", id, err)
 	}
 
-	log.Info("Peer has been removed from service", "id", id)
+	log.Info("Peer has been removed from service", "peer_id", id)
 	return nil
 }

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/cosmos/cosmos-sdk v0.47.17
 	github.com/gin-contrib/cors v1.7.6
 	github.com/gin-gonic/gin v1.10.1
-	github.com/sentinel-official/sentinel-go-sdk v1.0.0-rc.6.0.20250823130500-cadee865b3b9
+	github.com/sentinel-official/sentinel-go-sdk v1.0.0-rc.6.0.20250823134443-82024aa6258c
 	github.com/sentinel-official/sentinelhub/v12 v12.0.0-rc.11
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/pflag v1.0.7

go.sum

@@ -511,8 +511,8 @@ github.com/secure-io/siv-go v0.0.0-20180922214919-5ff40651e2c4 h1:zOjq+1/uLzn/Xo
 github.com/secure-io/siv-go v0.0.0-20180922214919-5ff40651e2c4/go.mod h1:aI+8yClBW+1uovkHw6HM01YXnYB8vohtB9C83wzx34E=
 github.com/seiflotfy/cuckoofilter v0.0.0-20220411075957-e3b120b3f5fb h1:XfLJSPIOUX+osiMraVgIrMR27uMXnRJWGm1+GL8/63U=
 github.com/seiflotfy/cuckoofilter v0.0.0-20220411075957-e3b120b3f5fb/go.mod h1:bR6DqgcAl1zTcOX8/pE2Qkj9XO00eCNqmKb7lXP8EAg=
-github.com/sentinel-official/sentinel-go-sdk v1.0.0-rc.6.0.20250823130500-cadee865b3b9 h1:lk44o2OQ2Y8dZkucsNe/0HWcsxkxLco27Xq4jZU408I=
-github.com/sentinel-official/sentinel-go-sdk v1.0.0-rc.6.0.20250823130500-cadee865b3b9/go.mod h1:jatyUZIc/7I7xmLI6fmbhzMjJ+si04XR2f0CksWtiXw=
+github.com/sentinel-official/sentinel-go-sdk v1.0.0-rc.6.0.20250823134443-82024aa6258c h1:x24atM/uRoF+1e5XeArHF8RVG8gBJssQRdoLphc4zcc=
+github.com/sentinel-official/sentinel-go-sdk v1.0.0-rc.6.0.20250823134443-82024aa6258c/go.mod h1:jatyUZIc/7I7xmLI6fmbhzMjJ+si04XR2f0CksWtiXw=
 github.com/sentinel-official/sentinelhub/v12 v12.0.0-rc.11 h1:B3ylnKZSLP9aDYnS2f98Gx/QTohPuD36tjl75kYJSMs=
 github.com/sentinel-official/sentinelhub/v12 v12.0.0-rc.11/go.mod h1:LXcOAEffAn4Pw58xhoYN9j8ZVuM02OCgPNTI6lkGG2E=
 github.com/shirou/gopsutil/v4 v4.25.7 h1:bNb2JuqKuAu3tRlPv5piSmBZyMfecwQ+t/ILq+1JqVM=

workers/geoip.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
+	logger "github.com/sentinel-official/sentinel-go-sdk/libs/log"
 
 	"github.com/sentinel-official/sentinel-dvpnx/core"
 )
@@ -15,19 +16,22 @@ const NameGeoIPLocation = "geoip_location"
 // NewGeoIPLocationWorker creates a worker to periodically update the GeoIP location in the context.
 // This worker fetches the GeoIP location and updates the context at regular intervals.
 func NewGeoIPLocationWorker(c *core.Context, interval time.Duration) cron.Worker {
+	log := logger.With("module", "workers", "name", NameGeoIPLocation)
+
 	// Handler function that fetches the GeoIP location and updates the context.
 	handlerFunc := func(ctx context.Context) error {
 		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 		defer cancel()
 
 		// Fetch the GeoIP location using the GeoIP client.
-		location, err := c.GeoIPClient().Get(ctx, "")
+		loc, err := c.GeoIPClient().Get(ctx, "")
 		if err != nil {
 			return fmt.Errorf("getting GeoIP location: %w", err)
 		}
 
 		// Update the context with the fetched location.
-		c.SetLocation(location)
+		log.Debug("Updating context", "city", loc.City, "country", loc.Country)
+		c.SetLocation(loc)
 
 		return nil
 	}

workers/rpc.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
+	logger "github.com/sentinel-official/sentinel-go-sdk/libs/log"
 
 	"github.com/sentinel-official/sentinel-dvpnx/core"
 )
@@ -20,6 +21,7 @@ const NameBestRPCAddr = "best_rpc_addr"
 // sorts them in ascending order of latency, and updates the context.
 func NewBestRPCAddrWorker(c *core.Context, interval time.Duration) cron.Worker {
 	client := &http.Client{Timeout: 5 * time.Second}
+	log := logger.With("module", "workers", "name", NameBestRPCAddr)
 
 	// Handler function that measures RPC address latencies and updates the context.
 	handlerFunc := func(ctx context.Context) error {
@@ -84,7 +86,7 @@ func NewBestRPCAddrWorker(c *core.Context, interval time.Duration) cron.Worker {
 			return latencies[addrs[i]] < latencies[addrs[j]]
 		})
 
-		// Update the context with the sorted list of RPC addresses.
+		log.Debug("Updating context", "addrs", addrs)
 		c.SetRPCAddrs(addrs)
 
 		return nil

workers/session.go

@@ -8,6 +8,7 @@ import (
 	"cosmossdk.io/math"
 	"github.com/cosmos/cosmos-sdk/types"
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
+	logger "github.com/sentinel-official/sentinel-go-sdk/libs/log"
 	"github.com/sentinel-official/sentinelhub/v12/types/v1"
 
 	"github.com/sentinel-official/sentinel-dvpnx/core"
@@ -25,6 +26,8 @@ const (
 // This worker retrieves session data from the database, validates it against the blockchain,
 // and broadcasts any updates as transactions.
 func NewSessionUsageSyncWithBlockchainWorker(c *core.Context, interval time.Duration) cron.Worker {
+	log := logger.With("module", "workers", "name", NameSessionUsageSyncWithBlockchain)
+
 	handlerFunc := func(ctx context.Context) error {
 		// Retrieve session records from the database.
 		query := map[string]interface{}{
@@ -44,14 +47,25 @@ func NewSessionUsageSyncWithBlockchainWorker(c *core.Context, interval time.Dura
 				return fmt.Errorf("querying session %d from blockchain: %w", item.GetID(), err)
 			}
 			if session == nil {
+				log.Debug("Skipping session",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "nil session",
+				)
 				continue
 			}
 			if session.GetUploadBytes().Equal(item.GetRxBytes()) {
+				log.Debug("Skipping session",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "already up-to-date",
+				)
 				continue
 			}
 
 			// Generate an update message for the session.
 			msg := item.MsgUpdateSessionRequest()
+			log.Debug("Adding session to update list",
+				"id", item.GetID(), "peer_id", item.GetPeerID(), "download_bytes", msg.DownloadBytes,
+				"duration", msg.Duration, "upload_bytes", msg.UploadBytes,
+			)
+
 			msgs = append(msgs, msg)
 		}
 
@@ -73,6 +87,8 @@ func NewSessionUsageSyncWithBlockchainWorker(c *core.Context, interval time.Dura
 // 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 *core.Context, interval time.Duration) cron.Worker {
+	log := logger.With("module", "workers", "name", NameSessionUsageSyncWithDatabase)
+
 	handlerFunc := func(ctx context.Context) error {
 		// Fetch peer usage statistics from the service.
 		items, err := c.Service().PeerStatistics()
@@ -81,14 +97,22 @@ func NewSessionUsageSyncWithDatabaseWorker(c *core.Context, interval time.Durati
 		}
 
 		// Update the database with the fetched statistics.
-		for id, item := range items {
+		for peerID, item := range items {
+			if time.Since(item.UpdatedAt) > interval {
+				log.Debug("Skipping session",
+					"id", 0, "peer_id", peerID, "cause", "already up-to-date",
+					"updated_at", item.UpdatedAt,
+				)
+				continue
+			}
+
 			// Convert usage statistics to strings for database storage.
 			rxBytes := math.NewInt(item.RxBytes).String()
 			txBytes := math.NewInt(item.TxBytes).String()
 
 			// Define query to find the session by peer id.
 			query := map[string]interface{}{
-				"peer_id": id,
+				"peer_id": peerID,
 			}
 
 			// Define updates to apply to the session record.
@@ -97,9 +121,11 @@ func NewSessionUsageSyncWithDatabaseWorker(c *core.Context, interval time.Durati
 				"tx_bytes": txBytes,
 			}
 
-			// Update the session in the database.
+			log.Debug("Updating session in database",
+				"id", 0, "peer_id", peerID, "rx_bytes", rxBytes, "tx_bytes", txBytes,
+			)
 			if _, err := operations.SessionFindOneAndUpdate(c.Database(), query, updates); err != nil {
-				return fmt.Errorf("updating session for peer %q in database: %w", id, err)
+				return fmt.Errorf("updating session for peer %q in database: %w", peerID, err)
 			}
 		}
 
@@ -115,6 +141,8 @@ func NewSessionUsageSyncWithDatabaseWorker(c *core.Context, interval time.Durati
 // 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 *core.Context, interval time.Duration) cron.Worker {
+	log := logger.With("module", "workers", "name", NameSessionUsageValidate)
+
 	handlerFunc := func(ctx context.Context) error {
 		// Retrieve session records from the database.
 		query := map[string]interface{}{
@@ -133,22 +161,35 @@ func NewSessionUsageValidateWorker(c *core.Context, interval time.Duration) cron
 			// Check if the session exceeds the maximum allowed bytes.
 			maxBytes := item.GetMaxBytes()
 			if !maxBytes.IsZero() && item.GetTotalBytes().GTE(maxBytes) {
+				log.Debug("Marking peer for removing from service",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "exceeds max bytes",
+					"total_bytes", item.GetTotalBytes(), "max_bytes", item.GetMaxBytes(),
+				)
 				removePeer = true
 			}
 
 			// Check if the session exceeds the maximum allowed duration.
 			maxDuration := item.GetMaxDuration()
 			if maxDuration != 0 && item.GetDuration() >= maxDuration {
+				log.Debug("Marking peer for removing from service",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "exceeds max duration",
+					"duration", item.GetDuration(), "max_duration", maxDuration,
+				)
 				removePeer = true
 			}
 
 			// Ensure that only sessions of the current service type are validated.
 			if item.GetServiceType() != c.Service().Type() {
+				log.Debug("Skipping peer",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "invalid service type",
+					"got", item.GetServiceType(), "expected", c.Service().Type(),
+				)
 				removePeer = false
 			}
 
 			// If the session exceeded any limits, remove the associated peer.
 			if removePeer {
+				log.Info("Removing peer from service", "id", item.GetID(), "peer_id", item.GetPeerID())
 				if err := c.RemovePeerIfExists(ctx, item.GetPeerID()); err != nil {
 					return fmt.Errorf("removing peer %q for session %d from service: %w", item.GetPeerID(), item.GetID(), err)
 				}
@@ -167,6 +208,8 @@ func NewSessionUsageValidateWorker(c *core.Context, interval time.Duration) cron
 // 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 *core.Context, interval time.Duration) cron.Worker {
+	log := logger.With("module", "workers", "name", NameSessionValidate)
+
 	handlerFunc := func(ctx context.Context) error {
 		// Retrieve session records from the database.
 		query := map[string]interface{}{
@@ -189,20 +232,32 @@ func NewSessionValidateWorker(c *core.Context, interval time.Duration) cron.Work
 
 			// Remove peer if the session is missing on the blockchain.
 			if session == nil {
+				log.Debug("Marking peer for removing from service",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "nil session",
+				)
 				removePeer = true
 			}
 			// Remove peer if the session status is not active.
 			if session != nil && !session.GetStatus().Equal(v1.StatusActive) {
+				log.Debug("Marking peer for removing from service",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "invalid session status",
+					"got", session.GetStatus(), "expected", v1.StatusActive,
+				)
 				removePeer = true
 			}
 
 			// Ensure that only sessions of the current service type are validated.
 			if item.GetServiceType() != c.Service().Type() {
+				log.Debug("Skipping peer",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "invalid service type",
+					"got", item.GetServiceType(), "expected", c.Service().Type(),
+				)
 				removePeer = false
 			}
 
 			// Remove the associated peer if validation fails.
 			if removePeer {
+				log.Info("Removing peer from service", "id", item.GetID(), "peer_id", item.GetPeerID())
 				if err := c.RemovePeerIfExists(ctx, item.GetPeerID()); err != nil {
 					return fmt.Errorf("removing peer %q for session %d from service: %w", item.GetPeerID(), item.GetID(), err)
 				}
@@ -212,6 +267,9 @@ func NewSessionValidateWorker(c *core.Context, interval time.Duration) cron.Work
 
 			// Delete session if the session is missing on the blockchain.
 			if session == nil {
+				log.Debug("Marking session for deleting from database",
+					"id", item.GetID(), "peer_id", item.GetPeerID(), "cause", "nil session",
+				)
 				deleteSession = true
 			}
 
@@ -221,6 +279,7 @@ func NewSessionValidateWorker(c *core.Context, interval time.Duration) cron.Work
 					"id": item.GetID(),
 				}
 
+				log.Info("Deleting session from database", "id", item.GetID(), "peer_id", item.GetPeerID())
 				if _, err := operations.SessionFindOneAndDelete(c.Database(), query); err != nil {
 					return fmt.Errorf("deleting session %d from database: %w", item.GetID(), err)
 				}

workers/speedtest.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/sentinel-official/sentinel-go-sdk/libs/cron"
+	logger "github.com/sentinel-official/sentinel-go-sdk/libs/log"
 	"github.com/sentinel-official/sentinel-go-sdk/libs/speedtest"
 
 	"github.com/sentinel-official/sentinel-dvpnx/core"
@@ -16,6 +17,8 @@ const NameSpeedtest = "speedtest"
 // NewSpeedtestWorker creates a worker that performs periodic speed tests and updates the context with the results.
 // This worker measures the download and upload speeds and sets the results in the application's context.
 func NewSpeedtestWorker(c *core.Context, interval time.Duration) cron.Worker {
+	log := logger.With("module", "workers", "name", NameSpeedtest)
+
 	// Handler function that performs the speed test and updates the context.
 	handlerFunc := func(ctx context.Context) error {
 		// Run the speed test to measure download and upload speeds.
@@ -24,7 +27,7 @@ func NewSpeedtestWorker(c *core.Context, interval time.Duration) cron.Worker {
 			return fmt.Errorf("running speed test: %w", err)
 		}
 
-		// Update the context with the obtained speed test results.
+		log.Debug("Updating context", "dl_speed", dlSpeed, "ul_speed", ulSpeed)
 		c.SetSpeedtestResults(dlSpeed, ulSpeed)
 
 		return nil