infra/ii/svc: 显式调用 ctx cancel

v4/infra/ii/svc/service.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"golib/v4/features/mo"
-	"golib/v4/gio"
 	"golib/v4/infra/ii"
 	"golib/v4/log"
 )
@@ -74,7 +73,9 @@ func (s *Service) FindAll(name ii.Name) ([]Row, error) {
 	}
 	opts := mo.Options.Find()
 	opts.SetSort(mo.NewSorter(ii.CreationTime, mo.SortDESC))
-	cursor, err := s.openColl(info).Find(gio.ContextTimeout(s.Timeout), mo.D{}, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor, err := s.openColl(info).Find(ctx, mo.D{}, opts)
 	if err != nil {
 		s.Log.Error("svc.FindAll.%s: %s", name, err)
 		return nil, errors.Join(ErrInternalError, err)
@@ -124,7 +125,9 @@ func (s *Service) FindWith(name ii.Name, filter, sort, project mo.Filter, skip,
 			opts.SetLimit(findLimitRows) // 如果没有过滤条件, 限制返回数量
 		}
 	}
-	cursor, err := s.openColl(info).Find(s.newContext(), query, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor, err := s.openColl(info).Find(ctx, query, opts)
 	if err != nil {
 		s.Log.Error("svc.Find.%s: %s filter: %v", name, err, query)
 		return errors.Join(ErrInternalError, err)
@@ -160,7 +163,9 @@ func (s *Service) FindOneWith(name ii.Name, filter, sort, project mo.Filter, v a
 	} else {
 		opts.SetSort(mo.NewSorter(ii.CreationTime, mo.SortDESC).Done())
 	}
-	cursor := s.openColl(info).FindOne(s.newContext(), query, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor := s.openColl(info).FindOne(ctx, query, opts)
 	if err := cursor.Err(); err != nil {
 		if errors.Is(err, mo.ErrNoDocuments) {
 			return err
@@ -187,7 +192,9 @@ func (s *Service) FindOneAndDelete(name ii.Name, filter mo.Filter) error {
 		s.Log.Error("svc.FindOneAndDelete.%s: QueryFilterCheck: %s filter: %v", name, err, query)
 		return errors.Join(ErrDataError, err)
 	}
-	result := s.openColl(info).FindOneAndDelete(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result := s.openColl(info).FindOneAndDelete(ctx, query)
 	if err := result.Err(); err != nil {
 		if IsErrNoDocuments(err) {
 			return err
@@ -208,7 +215,9 @@ func (s *Service) DeleteOne(name ii.Name, filter mo.Filter) error {
 		return ErrItemNotfound
 	}
 	query := filter.Done()
-	result, err := s.openColl(info).DeleteOne(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result, err := s.openColl(info).DeleteOne(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.DeleteOne.%s: %s filter: %v", name, err, query)
 		return errors.Join(ErrInternalError, err)
@@ -226,7 +235,9 @@ func (s *Service) DeleteMany(name ii.Name, filter mo.Filter) error {
 		return ErrItemNotfound
 	}
 	query := filter.Done()
-	result, err := s.openColl(info).DeleteMany(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result, err := s.openColl(info).DeleteMany(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.DeleteMany.%s: %s filter: %v", name, err, query)
 		return errors.Join(ErrInternalError, err)
@@ -250,7 +261,9 @@ func (s *Service) FindOneAndUpdate(name ii.Name, filter, updater mo.Filter) erro
 		return errors.Join(ErrDataError, err)
 	}
 	update := updater.Done()
-	result := s.openColl(info).FindOneAndUpdate(s.newContext(), query, update)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result := s.openColl(info).FindOneAndUpdate(ctx, query, update)
 	if err := result.Err(); err != nil {
 		if errors.Is(err, mo.ErrNoDocuments) {
 			return err
@@ -271,7 +284,9 @@ func (s *Service) EstimatedDocumentCount(name ii.Name) (int64, error) {
 		s.Log.Error("svc.EstimatedDocumentCount.%s: item not found", name)
 		return 0, ErrItemNotfound
 	}
-	length, err := s.openColl(info).EstimatedDocumentCount(s.newContext())
+	ctx, cancel := s.newContext()
+	defer cancel()
+	length, err := s.openColl(info).EstimatedDocumentCount(ctx)
 	if err != nil {
 		s.Log.Error("svc.EstimatedDocumentCount.%s: %s", name, err)
 		return 0, errors.Join(ErrInternalError, err)
@@ -291,7 +306,9 @@ func (s *Service) CountDocuments(name ii.Name, filter mo.Filter) (int64, error)
 		s.Log.Error("svc.CountDocuments.%s: QueryFilterCheck: %s filter: %v", name, err, query)
 		return 0, errors.Join(ErrDataError, err)
 	}
-	length, err := s.openColl(info).CountDocuments(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	length, err := s.openColl(info).CountDocuments(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.CountDocuments.%s: %s filter: %v", name, err, query)
 		return 0, errors.Join(ErrInternalError, err)
@@ -317,7 +334,9 @@ func (s *Service) InsertOne(name ii.Name, value any) (mo.ObjectID, error) {
 		s.Log.Error("svc.InsertOne.%s: PrepareInsert: %s", name, err)
 		return mo.NilObjectID, errors.Join(ErrDataError, err)
 	}
-	result, err := s.openColl(info).InsertOne(s.newContext(), doc)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result, err := s.openColl(info).InsertOne(ctx, doc)
 	if err != nil {
 		s.Log.Error("svc.InsertOne.%s: %s data: %v", name, err, doc)
 		return mo.NilObjectID, errors.Join(ErrInternalError, err)
@@ -348,7 +367,9 @@ func (s *Service) InsertMany(name ii.Name, value any) (mo.A, error) {
 		s.Log.Error("svc.InsertMany.%s: toMaps: %s", name, err)
 		return nil, errors.Join(ErrDataError, err)
 	}
-	result, err := s.openColl(info).InsertMany(s.newContext(), docs)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result, err := s.openColl(info).InsertMany(ctx, docs)
 	if err != nil {
 		s.Log.Error("svc.InsertMany.%s: %s", name, err)
 		return nil, errors.Join(ErrInternalError, err)
@@ -384,7 +405,9 @@ func (s *Service) UpdateOne(name ii.Name, filter, updater mo.Filter) error {
 	_, upsert := mo.HasOptIn(update, mo.OptSetOnInsert)
 	opts.SetUpsert(upsert)
 
-	_, err := s.openColl(info).UpdateOne(s.newContext(), query, update, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	_, err := s.openColl(info).UpdateOne(ctx, query, update, opts)
 	if err != nil {
 		s.Log.Error("svc.UpdateOne.%s: %s filter: %v updater: %v", name, err, filter, update)
 		return errors.Join(ErrInternalError, err)
@@ -425,7 +448,9 @@ func (s *Service) UpdateMany(name ii.Name, filter, updater mo.Filter) error {
 	_, upsert := mo.HasOptIn(update, mo.OptSetOnInsert)
 	opts.SetUpsert(upsert)
 
-	result, err := s.openColl(info).UpdateMany(s.newContext(), query, update, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	result, err := s.openColl(info).UpdateMany(ctx, query, update, opts)
 	if err != nil {
 		s.Log.Error("svc.UpdateMany.%s: %s filter: %v updater: %v", name, err, query, update)
 		return errors.Join(ErrInternalError, err)
@@ -465,7 +490,9 @@ func (s *Service) Aggregate(name ii.Name, pipe mo.Pipeline, v any) error {
 		stage, lookup = s.Cache.SpitPipe(info, pipe)
 	}
 
-	cursor, err := s.openColl(info).Aggregate(s.newContext(), stage)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor, err := s.openColl(info).Aggregate(ctx, stage)
 	if err != nil {
 		s.Log.Error("svc.Aggregate.%s: %s pipe: %v", name, err, pipe)
 		return errors.Join(ErrInternalError, err)
@@ -514,12 +541,15 @@ func (s *Service) refreshCache(info *ii.ItemInfo) {
 func (s *Service) handleRefresh() {
 	for info := range s.refreshCh {
 		qt := time.Now()
-		cursor, err := s.openColl(info).Find(s.newContext(), mo.D{})
+		ctx, cancel := s.newContext()
+		cursor, err := s.openColl(info).Find(ctx, mo.D{})
 		if err != nil {
+			cancel()
 			s.Cache.Clear(info.Name) // 查询失败时则清除内存缓存, 防止信息不一致
 			s.Log.Error("svc.refreshCache: %s->%s", info.Name, err)
 			continue
 		}
+		cancel()
 		qts := time.Now().Sub(qt)
 
 		dt := time.Now()
@@ -545,14 +575,15 @@ func (s *Service) openColl(info *ii.ItemInfo) *mo.Collection {
 	return s.Client.Database(info.Name.Database()).Collection(info.Name.Collection())
 }
 
-func (s *Service) newContext() context.Context {
-	root := gio.ContextTimeout(s.Timeout)
+func (s *Service) newContext() (context.Context, context.CancelFunc) {
+	root, cancel := context.WithTimeout(context.Background(), s.Timeout)
 	go func(s *Service) {
 		d := time.Duration(s.Timeout.Seconds()*0.8) * time.Second
-		ctx, cancel := context.WithTimeout(root, d)
-		defer cancel()
-		<-ctx.Done()
-		s.Log.Warn("svc.HealthCheck: Warning: Exec has been running for %f seconds.", d.Seconds())
+		select {
+		case <-time.After(d):
+			s.Log.Warn("svc.HealthCheck: Warning: Exec has been running for %f seconds.", d.Seconds())
+		case <-root.Done():
+		}
 	}(s)
-	return root
+	return root, cancel
 }

v4/infra/ii/svc/svc.go

@@ -51,7 +51,9 @@ func (s *WithUser) FindWith(name ii.Name, filter mo.Filter, v any) error {
 		return errors.Join(ErrPermissionDenied, err)
 	}
 
-	cursor, err := s.openColl(info).Find(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor, err := s.openColl(info).Find(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.Find.%s: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return errors.Join(ErrInternalError, err)
@@ -81,7 +83,9 @@ func (s *WithUser) FindOneWith(name ii.Name, filter mo.Filter, v any) error {
 		return ErrPermissionDenied
 	}
 
-	cursor := s.openColl(info).FindOne(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor := s.openColl(info).FindOne(ctx, query)
 	if err := cursor.Err(); err != nil {
 		if errors.Is(err, mo.ErrNoDocuments) {
 			return err
@@ -113,7 +117,9 @@ func (s *WithUser) FindOneAndDelete(name ii.Name, filter mo.Filter) error {
 		s.Log.Error("svc.FindOneAndDelete.%s: setAC: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return ErrPermissionDenied
 	}
-	ret := s.openColl(info).FindOneAndDelete(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret := s.openColl(info).FindOneAndDelete(ctx, query)
 	if err := ret.Err(); err != nil {
 		if errors.Is(err, mo.ErrNoDocuments) {
 			return err
@@ -138,8 +144,9 @@ func (s *WithUser) DeleteOne(name ii.Name, filter mo.Filter) error {
 		s.Log.Error("svc.DeleteOne.%s: setAC: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return ErrPermissionDenied
 	}
-
-	ret, err := s.openColl(info).DeleteOne(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret, err := s.openColl(info).DeleteOne(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.DeleteOne.%s: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return err
@@ -161,8 +168,9 @@ func (s *WithUser) DeleteMany(name ii.Name, filter mo.Filter) error {
 		s.Log.Error("svc.DeleteMany.%s: setAC: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return ErrPermissionDenied
 	}
-
-	ret, err := s.openColl(info).DeleteMany(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret, err := s.openColl(info).DeleteMany(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.DeleteMany.%s: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return err
@@ -194,7 +202,9 @@ func (s *WithUser) FindOneAndUpdate(name ii.Name, filter, updater mo.Filter) err
 		s.Log.Error("svc.FindOneAndUpdate.%s: setAC: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return ErrPermissionDenied
 	}
-	ret := s.openColl(info).FindOneAndUpdate(s.newContext(), query, update)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret := s.openColl(info).FindOneAndUpdate(ctx, query, update)
 	if err := ret.Err(); err != nil {
 		if errors.Is(err, mo.ErrNoDocuments) {
 			return err
@@ -227,10 +237,12 @@ func (s *WithUser) EstimatedDocumentCount(name ii.Name) (int64, error) {
 		err    error
 	)
 
+	ctx, cancel := s.newContext()
+	defer cancel()
 	if len(filter) > 0 {
-		length, err = s.openColl(info).CountDocuments(s.newContext(), filter)
+		length, err = s.openColl(info).CountDocuments(ctx, filter)
 	} else {
-		length, err = s.openColl(info).EstimatedDocumentCount(s.newContext())
+		length, err = s.openColl(info).EstimatedDocumentCount(ctx)
 	}
 
 	if err != nil {
@@ -257,7 +269,9 @@ func (s *WithUser) CountDocuments(name ii.Name, filter mo.Filter) (int64, error)
 		s.Log.Error("svc.CountDocuments.%s: setAC: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return 0, ErrPermissionDenied
 	}
-	length, err := s.openColl(info).CountDocuments(s.newContext(), query)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	length, err := s.openColl(info).CountDocuments(ctx, query)
 	if err != nil {
 		s.Log.Error("svc.CountDocuments.%s: %s filter: %v UID: %s", name, err, query, s.User.ID().Hex())
 		return 0, ErrInternalError
@@ -279,8 +293,9 @@ func (s *WithUser) InsertOne(name ii.Name, doc mo.M) (mo.ObjectID, error) {
 		s.Log.Error("svc.InsertOne.%s: %s data: %v UID: %s", name, err, doc, s.User.ID().Hex())
 		return mo.NilObjectID, ErrDataError
 	}
-
-	ret, err := s.openColl(info).InsertOne(s.newContext(), doc)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret, err := s.openColl(info).InsertOne(ctx, doc)
 	if err != nil {
 		s.Log.Error("svc.InsertOne.%s: %s data: %v UID: %s", name, err, doc, s.User.ID().Hex())
 		return mo.NilObjectID, ErrInternalError
@@ -312,7 +327,9 @@ func (s *WithUser) InsertMany(name ii.Name, value any) (mo.A, error) {
 		s.Log.Error("svc.InsertMany.%s: %s UID: %s", name, err, s.User.ID().Hex())
 		return nil, ErrDataError
 	}
-	ret, err := s.openColl(info).InsertMany(s.newContext(), docs)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret, err := s.openColl(info).InsertMany(ctx, docs)
 	if err != nil {
 		s.Log.Error("svc.InsertMany.%s: %s UID: %s", name, err, s.User.ID().Hex())
 		return nil, ErrInternalError
@@ -349,7 +366,9 @@ func (s *WithUser) UpdateOne(name ii.Name, filter, updater mo.Filter) error {
 	_, upsert := mo.HasOptIn(update, mo.OptSetOnInsert)
 	opts.SetUpsert(upsert)
 
-	ret, err := s.openColl(info).UpdateOne(s.newContext(), query, update, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret, err := s.openColl(info).UpdateOne(ctx, query, update, opts)
 	if err != nil {
 		s.Log.Error("svc.UpdateOne.%s: %s filter: %v updater: %v UID: %s", name, err, query, update, s.User.ID().Hex())
 		return ErrInternalError
@@ -397,7 +416,9 @@ func (s *WithUser) UpdateMany(name ii.Name, filter, updater mo.Filter) error {
 	_, upsert := mo.HasOptIn(update, mo.OptSetOnInsert)
 	opts.SetUpsert(upsert)
 
-	ret, err := s.openColl(info).UpdateMany(s.newContext(), filter, update, opts)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	ret, err := s.openColl(info).UpdateMany(ctx, filter, update, opts)
 	if err != nil {
 		s.Log.Error("svc.UpdateMany.%s: %s filter: %v updater: %v UID: %s", name, err, filter, update, s.User.ID().Hex())
 		return ErrInternalError
@@ -451,7 +472,9 @@ func (s *WithUser) Aggregate(name ii.Name, pipe mo.Pipeline, v any) error {
 		stage, lookup = s.Cache.SpitPipe(info, pipe)
 	}
 
-	cursor, err := s.openColl(info).Aggregate(s.newContext(), stage)
+	ctx, cancel := s.newContext()
+	defer cancel()
+	cursor, err := s.openColl(info).Aggregate(ctx, stage)
 	if err != nil {
 		s.Log.Error("svc.Aggregate.%s: %s pipe: %v UID: %s", name, err, pipe, s.User.ID().Hex())
 		return ErrInternalError