batch output insert

[mgukov]

• feat(be): batch task output insert
• feat(be): batch task output insert, boltDB
• feat(be): batch task output insert, boltDB
• feat(be): batch task output insert, boltDB

[Copilot]

Pull Request Overview

This PR implements batch processing for task output inserts to improve performance by reducing database operations. Instead of inserting task outputs one by one, the system now collects outputs in batches and flushes them either when reaching a batch size limit or at regular intervals.

• Introduces batching mechanism with configurable batch size and flush interval
• Adds new database methods for batch insertion in both SQL and BoltDB implementations
• Updates task output structure to include stage ID for better tracking

Reviewed Changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
services/tasks/TaskPool.go	Implements batch collection and flushing logic with ticker-based intervals
db/sql/task.go	Adds SQL batch insert method using bulk insert query
db/bolt/task.go	Adds BoltDB batch insert method using transaction
db/Task.go	Adds StageID field to TaskOutput struct
db/Store.go	Adds batch insert method to TaskManager interface
db/sql/migrations/v2.16.2.sql	Adds stage_id column to task__output table

[Copilot]

The newOutput variable is being modified after assignment to record.task.currentOutput. The StageID is set on line 275, but this modification won't be reflected in the task's currentOutput since it's a copy, not a reference to the same object.

[Copilot]

This infinite loop has no exit condition and will prevent the goroutine from terminating gracefully. Consider adding a context or done channel to allow proper shutdown.

[Copilot]

The ticker is never stopped, which can lead to a goroutine leak. Consider adding defer logTicker.Stop() or stopping it when the loop exits.

Suggested change

	logTicker := time.NewTicker(TaskOutputInsertIntervalMs * time.Millisecond)
	logTicker := time.NewTicker(TaskOutputInsertIntervalMs * time.Millisecond)
	defer logTicker.Stop()

[Copilot]

Creating a new slice instead of clearing the existing one may cause unnecessary allocations. Consider using *logs = (*logs)[:0] to reuse the underlying array.

Suggested change

	*logs = make([]logRecord, 0)
	*logs = (*logs)[:0]

[fiftin]

@cursor review

[cursor]

Bugbot free trial expires on August 10, 2025
Learn more in the Cursor dashboard.

[cursor]

Bug

The new batching logic for task output introduces several critical issues:

• Missing Database IDs: Stage parsing via MoveToNextStage now receives TaskOutput objects without database IDs. Previously, CreateTaskOutput would assign an ID before stage processing. The new batch insert (InsertTaskOutputBatch) occurs later and does not update the original objects with their generated IDs, breaking logic dependent on valid output IDs.
• Incorrect Previous Output: Within the batch processing loop, record.task.currentOutput is incorrectly updated. Since record.task is the same TaskRunner instance for all records from a task, currentOutput := record.task.currentOutput retrieves the output from the previous iteration within the batch, not the actual previous output for the current log record. This breaks the stage progression logic that relies on the correct preceding output.
• Inconsistent State on Failure: Task state (task.currentState, task.currentStage) is updated by MoveToNextStage before the batch of outputs is persisted. If the batch insert fails, the in-memory task state becomes inconsistent with the database.

Locations (3)

• services/tasks/TaskPool.go#L235-L288
• db/sql/task.go#L231-L252
• db/bolt/task.go#L129-L135

Fix in Cursor • Fix in Web