mazyar/tm_back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

worker concurrency config

Browse Source
This commit is contained in:
Mazyar
2026-05-12 23:56:10 +03:30
parent 5e8d4f67b2
commit be53fa8054
3 changed files with 51 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/worker/bootstrap/bootstrap.go
+12 -6
View File
@@ -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()
}()
cmd/worker/bootstrap/config.go
+11 -7
View File
@@ -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.
cmd/worker/workers/notice.go
+28 -7
View File
@@ -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
}