fix(downloader): serialize attempt ledger updates

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)

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()