diff --git a/cmd/internal/downloader/task_runner.go b/cmd/internal/downloader/task_runner.go index e53c777a..aa8b1302 100644 --- a/cmd/internal/downloader/task_runner.go +++ b/cmd/internal/downloader/task_runner.go @@ -46,6 +46,7 @@ type TaskRunner struct { attempts map[string]int wg sync.WaitGroup mu sync.Mutex + attemptMu sync.Mutex } type transferSpeeds struct { @@ -446,6 +447,9 @@ func (w *TaskRunner) resetTaskForAttempt(ctx context.Context, task client.Downlo w.setMemoryAttempt(task.ID, attempt) return nil } + w.attemptMu.Lock() + defer w.attemptMu.Unlock() + ledger, err := loadAttemptLedger(w.cfg.StateDir) if err != nil { return fmt.Errorf("load attempt ledger: %w", err) diff --git a/cmd/internal/downloader/task_runner_test.go b/cmd/internal/downloader/task_runner_test.go index 42d49b8b..dc3bee01 100644 --- a/cmd/internal/downloader/task_runner_test.go +++ b/cmd/internal/downloader/task_runner_test.go @@ -12,6 +12,7 @@ import ( "reflect" "strconv" "strings" + "sync" "testing" "time" @@ -1368,6 +1369,42 @@ func TestResetTaskForRestartAttemptSkipsAlreadyRecordedAttempt(t *testing.T) { } } +func TestResetTaskForAttemptSerializesLedgerUpdates(t *testing.T) { + const taskCount = 64 + + stateDir := t.TempDir() + w := NewTaskRunnerWithAPI(config.Config{StateDir: stateDir}, nil) + start := make(chan struct{}) + errs := make(chan error, taskCount) + var wg sync.WaitGroup + for i := range taskCount { + wg.Add(1) + go func() { + defer wg.Done() + <-start + task := clientTaskWithStatus("task-"+strconv.Itoa(i), "assigned") + errs <- w.resetTaskForAttempt(context.Background(), task, w.logger) + }() + } + + close(start) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + + ledger, err := loadAttemptLedger(stateDir) + if err != nil { + t.Fatal(err) + } + if len(ledger.Attempts) != taskCount { + t.Fatalf("expected %d attempt records, got %d", taskCount, len(ledger.Attempts)) + } +} + func TestRetainSeedKeepsDownloadedResult(t *testing.T) { dir := t.TempDir() stateDir := t.TempDir()