addLast adds a new tail element. It implements the merging logic for HookRun tasks by scanning the whole queue.
(tasks ...task.Task)
| 325 | // addLast adds a new tail element. |
| 326 | // It implements the merging logic for HookRun tasks by scanning the whole queue. |
| 327 | func (q *TaskQueue) AddLast(tasks ...task.Task) { |
| 328 | defer q.MeasureActionTime("AddLast")() |
| 329 | |
| 330 | for _, t := range tasks { |
| 331 | q.lazydebug("adding task to queue", func() []any { |
| 332 | return []any{ |
| 333 | slog.String(pkg.LogKeyQueue, q.Name), |
| 334 | slog.String(pkg.LogKeyTaskID, t.GetId()), |
| 335 | slog.String(pkg.LogKeyTaskType, string(t.GetType())), |
| 336 | slog.String(pkg.LogKeyTaskDescription, t.GetDescription()), |
| 337 | slog.Int(pkg.LogKeyQueueLengthBefore, q.storage.Length()), |
| 338 | } |
| 339 | }) |
| 340 | |
| 341 | if q.storage.Get(t.GetId()) != nil { |
| 342 | q.logger.Warn("task collision detected, unexpected behavior possible", slog.String(pkg.LogKeyQueue, q.Name), slog.String(pkg.LogKeyTaskID, t.GetId())) |
| 343 | } |
| 344 | |
| 345 | q.storage.AddLast(t) |
| 346 | q.queueTasksCounter.Add(t) |
| 347 | |
| 348 | // TODO: skip compactable check if is already compactable |
| 349 | |
| 350 | taskType := t.GetType() |
| 351 | if _, ok := q.compactableTypes[taskType]; ok { |
| 352 | q.lazydebug("task is mergeable, marking queue as dirty", func() []any { |
| 353 | return []any{ |
| 354 | slog.String(pkg.LogKeyQueue, q.Name), |
| 355 | slog.String(pkg.LogKeyTaskID, t.GetId()), |
| 356 | slog.String(pkg.LogKeyTaskType, string(taskType)), |
| 357 | slog.Int(pkg.LogKeyQueueLength, q.storage.Length()), |
| 358 | slog.Bool(pkg.LogKeyQueueIsDirty, q.queueTasksCounter.IsAnyCapReached()), |
| 359 | } |
| 360 | }) |
| 361 | |
| 362 | // Only trigger compaction if queue is getting long and we have mergeable tasks |
| 363 | if q.storage.Length() > compactionThreshold && q.queueTasksCounter.IsAnyCapReached() { |
| 364 | q.lazydebug("triggering compaction due to queue length", func() []any { |
| 365 | return []any{ |
| 366 | slog.String(pkg.LogKeyQueue, q.Name), |
| 367 | slog.Int(pkg.LogKeyQueueLength, q.storage.Length()), |
| 368 | slog.Int(pkg.LogKeyCompactionThreshold, compactionThreshold), |
| 369 | } |
| 370 | }) |
| 371 | |
| 372 | currentQueue := q.storage.Length() |
| 373 | q.compaction(q.queueTasksCounter.GetReachedCap()) |
| 374 | |
| 375 | q.lazydebug("compaction finished", func() []any { |
| 376 | return []any{ |
| 377 | slog.String(pkg.LogKeyQueue, q.Name), |
| 378 | slog.Int(pkg.LogKeyQueueLengthBefore, currentQueue), |
| 379 | slog.Int(pkg.LogKeyQueueLengthAfter, q.storage.Length()), |
| 380 | } |
| 381 | }) |
| 382 | } |
| 383 | } else { |
| 384 | q.lazydebug("task is not mergeable", func() []any { |
nothing calls this directly
no test coverage detected