fix: unexpected new db connections

interfaces/result_service.go

@@ -2,11 +2,14 @@ package interfaces
 
 import (
 	"github.com/crawlab-team/crawlab-db/generic"
+	"time"
 )
 
 type ResultService interface {
 	Insert(records ...interface{}) (err error)
 	List(query generic.ListQuery, opts *generic.ListOptions) (results []interface{}, err error)
 	Count(query generic.ListQuery) (n int, err error)
 	Index(fields []string)
+	SetTime(t time.Time)
+	GetTime() (t time.Time)
 }

result/service.go

@@ -2,14 +2,13 @@ package result
 
 import (
 	"fmt"
-	"github.com/crawlab-team/crawlab-core/entity"
 	"github.com/crawlab-team/crawlab-core/errors"
 	"github.com/crawlab-team/crawlab-core/interfaces"
 	"github.com/crawlab-team/crawlab-core/models/models"
 	"github.com/crawlab-team/crawlab-core/models/service"
 	"github.com/crawlab-team/go-trace"
 	"go.mongodb.org/mongo-driver/bson/primitive"
-	"time"
+	"sync"
 )
 
 func NewResultService(registryKey string, s *models.Spider) (svc2 interfaces.ResultService, err error) {
@@ -37,7 +36,7 @@ func NewResultService(registryKey string, s *models.Spider) (svc2 interfaces.Res
 	return svc, nil
 }
 
-var store = entity.NewTTLMap(5 * time.Second)
+var store = sync.Map{}
 
 func GetResultService(spiderId primitive.ObjectID, opts ...Option) (svc2 interfaces.ResultService, err error) {
 	// model service
@@ -62,7 +61,7 @@ func GetResultService(spiderId primitive.ObjectID, opts ...Option) (svc2 interfa
 	storeKey := s.ColId.Hex() + ":" + s.DataSourceId.Hex()
 
 	// attempt to load result service from store
-	res := store.Load(storeKey)
+	res, _ := store.Load(storeKey)
 	if res != nil {
 		svc, ok := res.(interfaces.ResultService)
 		if ok {

result/service_mongo.go

@@ -13,6 +13,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	mongo2 "go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"time"
 )
 
 type ServiceMongo struct {
@@ -23,6 +24,7 @@ type ServiceMongo struct {
 	// internals
 	colId primitive.ObjectID     // _id of models.DataCollection
 	dc    *models.DataCollection // models.DataCollection
+	t     time.Time
 }
 
 func (svc *ServiceMongo) List(query generic.ListQuery, opts *generic.ListOptions) (results []interface{}, err error) {
@@ -84,6 +86,14 @@ func (svc *ServiceMongo) Index(fields []string) {
 	}
 }
 
+func (svc *ServiceMongo) SetTime(t time.Time) {
+	svc.t = t
+}
+
+func (svc *ServiceMongo) GetTime() (t time.Time) {
+	return svc.t
+}
+
 func (svc *ServiceMongo) getList(query bson.M, opts *mongo.FindOptions) (results []interface{}, err error) {
 	list, err := svc.modelColSvc.GetList(query, opts)
 	if err != nil {
@@ -110,6 +120,7 @@ func NewResultServiceMongo(colId primitive.ObjectID, _ primitive.ObjectID) (svc2
 	// service
 	svc := &ServiceMongo{
 		colId: colId,
+		t:     time.Now(),
 	}
 
 	// dependency injection

task/stats/service.go

@@ -2,7 +2,6 @@ package stats
 
 import (
 	config2 "github.com/crawlab-team/crawlab-core/config"
-	"github.com/crawlab-team/crawlab-core/entity"
 	"github.com/crawlab-team/crawlab-core/interfaces"
 	"github.com/crawlab-team/crawlab-core/models/service"
 	"github.com/crawlab-team/crawlab-core/node/config"
@@ -26,7 +25,8 @@ type Service struct {
 
 	// internals
 	mu             sync.Mutex
-	resultServices *entity.TTLMap
+	resultServices sync.Map
+	rsTtl          time.Duration
 	logDriver      log.Driver
 }
 
@@ -47,11 +47,16 @@ func (svc *Service) InsertLogs(id primitive.ObjectID, logs ...string) (err error
 }
 
 func (svc *Service) getResultService(id primitive.ObjectID) (resultSvc interfaces.ResultService, err error) {
+	// atomic operation
+	svc.mu.Lock()
+	defer svc.mu.Unlock()
+
 	// attempt to get from cache
-	res := svc.resultServices.Load(id.Hex())
+	res, _ := svc.resultServices.Load(id.Hex())
 	if res != nil {
 		// hit in cache
 		resultSvc, ok := res.(interfaces.ResultService)
+		resultSvc.SetTime(time.Now())
 		if ok {
 			return resultSvc, nil
 		}
@@ -83,6 +88,25 @@ func (svc *Service) updateTaskStats(id primitive.ObjectID, resultCount int) {
 	})
 }
 
+func (svc *Service) cleanup() {
+	for {
+		// atomic operation
+		svc.mu.Lock()
+
+		svc.resultServices.Range(func(key, value interface{}) bool {
+			rs := value.(interfaces.ResultService)
+			if time.Now().After(rs.GetTime().Add(svc.rsTtl)) {
+				svc.resultServices.Delete(key)
+			}
+			return true
+		})
+
+		svc.mu.Unlock()
+
+		time.Sleep(10 * time.Minute)
+	}
+}
+
 func NewTaskStatsService(opts ...Option) (svc2 interfaces.TaskStatsService, err error) {
 	// base service
 	baseSvc, err := task.NewBaseService()
@@ -92,8 +116,9 @@ func NewTaskStatsService(opts ...Option) (svc2 interfaces.TaskStatsService, err
 
 	// service
 	svc := &Service{
+		mu:              sync.Mutex{},
 		TaskBaseService: baseSvc,
-		resultServices:  entity.NewTTLMap(5 * time.Second),
+		resultServices:  sync.Map{},
 	}
 
 	// apply options