From be53fa8054551fb3ae0a9a889af9826d66680459 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Tue, 12 May 2026 23:56:10 +0330 Subject: [PATCH] worker concurrency config --- cmd/worker/bootstrap/bootstrap.go | 18 ++++++++++------ cmd/worker/bootstrap/config.go | 18 +++++++++------- cmd/worker/workers/notice.go | 35 ++++++++++++++++++++++++------- 3 files changed, 51 insertions(+), 20 deletions(-) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 266f905..9e6befb 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -80,12 +80,14 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ - "worker_interval": config.Worker.Interval, - "notice_concurrency": config.Worker.NoticeWorkerConcurrency, - "notice_fetch_batch": config.Worker.NoticeFetchBatchSize, - "notice_processing_limit": config.Worker.NoticeProcessingLimit, - "queue_enabled": config.Queue.Enabled, - "queue_mode": config.Queue.Mode, + "worker_interval": config.Worker.Interval, + "notice_concurrency": config.Worker.NoticeWorkerConcurrency, + "notice_fetch_batch": config.Worker.NoticeFetchBatchSize, + "notice_processing_limit": config.Worker.NoticeProcessingLimit, + "notice_batch_pause": config.Worker.NoticeBatchPause.String(), + "notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(), + "queue_enabled": config.Queue.Enabled, + "queue_mode": config.Queue.Mode, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -182,6 +184,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger config.Worker.NoticeWorkerConcurrency, config.Worker.NoticeFetchBatchSize, config.Worker.DeleteProcessedNotices, + config.Worker.NoticeBatchPause, + config.Worker.NoticeFetchErrorBackoff, ) worker.Run() }, @@ -227,6 +231,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger config.Worker.NoticeWorkerConcurrency, config.Worker.NoticeFetchBatchSize, config.Worker.DeleteProcessedNotices, + config.Worker.NoticeBatchPause, + config.Worker.NoticeFetchErrorBackoff, ) w.Run() }() diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 65fbcbd..e9addc7 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -35,13 +35,17 @@ type QueueConfig struct { } type WorkerConfig struct { - Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` - NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` - DeleteProcessedNotices bool `env:"WORKER_DELETE_PROCESSED_NOTICES" envDefault:"true"` - NoticeWorkerConcurrency int `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"8"` - NoticeFetchBatchSize int `env:"WORKER_NOTICE_FETCH_BATCH" envDefault:"50"` - TranslationInterval string `env:"WORKER_TRANSLATION_INTERVAL" envDefault:"0 */30 * * * *"` - TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"` + Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` + NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` + DeleteProcessedNotices bool `env:"WORKER_DELETE_PROCESSED_NOTICES" envDefault:"true"` + NoticeWorkerConcurrency int `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"8"` + NoticeFetchBatchSize int `env:"WORKER_NOTICE_FETCH_BATCH" envDefault:"50"` + // NoticeBatchPause is slept after each notice batch finishes, before loading the next batch (0 = no pause). + NoticeBatchPause time.Duration `env:"WORKER_NOTICE_BATCH_PAUSE" envDefault:"0s"` + // NoticeFetchErrorBackoff is slept after a failed notice fetch before retrying the loop (0 = retry immediately). + NoticeFetchErrorBackoff time.Duration `env:"WORKER_NOTICE_FETCH_ERROR_BACKOFF" envDefault:"2s"` + TranslationInterval string `env:"WORKER_TRANSLATION_INTERVAL" envDefault:"0 */30 * * * *"` + TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"` } // AISummarizerConfig holds configuration for the external AI summarizer service. diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index 4830451..6846017 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -43,6 +43,10 @@ type NoticeWorker struct { Concurrency int // FetchBatchSize is how many unprocessed notices to load from MongoDB per batch (default 50). FetchBatchSize int + // BatchPause is slept after each batch completes, before fetching the next batch (0 = disabled). + BatchPause time.Duration + // FetchErrorBackoff is slept after a failed notice fetch before retrying (0 = disabled). + FetchErrorBackoff time.Duration // tenderIDSeedMu / tenderIDSeeded track which (sourceCode+year) counters have been seeded // from the current tenders collection during this process, so we only run the seeding @@ -51,7 +55,7 @@ type NoticeWorker struct { tenderIDSeeded map[string]bool } -func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, processingLimit, concurrency, fetchBatchSize int, deleteProcessedNotices bool) *NoticeWorker { +func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, processingLimit, concurrency, fetchBatchSize int, deleteProcessedNotices bool, batchPause, fetchErrorBackoff time.Duration) *NoticeWorker { if processingLimit < 0 { processingLimit = 5 // Default limit } @@ -61,6 +65,12 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif if fetchBatchSize < 1 { fetchBatchSize = 50 } + if batchPause < 0 { + batchPause = 0 + } + if fetchErrorBackoff < 0 { + fetchErrorBackoff = 0 + } return &NoticeWorker{ Mongo: mongo, Logger: logger, @@ -71,6 +81,8 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif DeleteProcessedNotices: deleteProcessedNotices, Concurrency: concurrency, FetchBatchSize: fetchBatchSize, + BatchPause: batchPause, + FetchErrorBackoff: fetchErrorBackoff, tenderIDSeeded: make(map[string]bool), } } @@ -130,9 +142,11 @@ func (w *NoticeWorker) cleanupProcessedNotices() { func (w *NoticeWorker) Run() { w.Logger.Info("Notice worker started", map[string]interface{}{ - "concurrency": w.Concurrency, - "fetch_batch": w.FetchBatchSize, - "processing_limit": w.ProcessingLimit, + "concurrency": w.Concurrency, + "fetch_batch": w.FetchBatchSize, + "processing_limit": w.ProcessingLimit, + "batch_pause": w.BatchPause.String(), + "fetch_error_backoff": w.FetchErrorBackoff.String(), }) var processedTotal atomic.Int64 @@ -155,6 +169,9 @@ func (w *NoticeWorker) Run() { w.Logger.Error("Failed to get notices", map[string]interface{}{ "error": err.Error(), }) + if w.FetchErrorBackoff > 0 { + time.Sleep(w.FetchErrorBackoff) + } continue } @@ -224,6 +241,10 @@ func (w *NoticeWorker) Run() { }) } + if w.BatchPause > 0 && len(notices) > 0 { + time.Sleep(w.BatchPause) + } + if maxToProcess > 0 && int(processedTotal.Load()) >= maxToProcess { w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ "processed_count": processedTotal.Load(), @@ -909,9 +930,9 @@ func (w *NoticeWorker) ensureTenderIDCounterSeeded(ctx context.Context, sourceCo w.tenderIDSeeded[key] = true w.Logger.Info("Tender id counter seeded", map[string]interface{}{ - "key": key, - "seed_value": currentMax, - "next_id": fmt.Sprintf("%s%d", key, currentMax+1), + "key": key, + "seed_value": currentMax, + "next_id": fmt.Sprintf("%s%d", key, currentMax+1), }) return nil }