core/txpool/blobpool: serialize legacy data conversions (#35387)

`conversionQueue` is meant to run billy/legacy migrations serially on a
single worker. On each `startConversion`, it currently always calls
`runNextTask()`, which overwrites `taskDone` and starts another
goroutine even when a migration is already running.

That breaks two invariants during an upgrade with legacy entries in both
the main store and limbo:

1. Migrations intended to be serial can run concurrently.
2. `close()` only waits for the most recently launched task, so shutdown
can close the stores while a forgotten migration is still using them.

Only start the next queued migration when no task is active (`taskDone
== nil`). The existing `taskDone` completion path then advances the
queue, and `close` reliably waits for the active task.

core/txpool/blobpool/conversion.go

@@ -164,7 +164,9 @@ func (q *conversionQueue) loop() {
 
 		case fn := <-q.startConversion:
 			q.queue = append(q.queue, fn)
-			q.runNextTask()
+			if q.taskDone == nil {
+				q.runNextTask()
+			}
 
 		case <-q.taskDone:
 			q.runNextTask()

core/txpool/blobpool/conversion_test.go

@@ -65,6 +65,71 @@ func TestConversionQueueDoubleClose(t *testing.T) {
 	queue.close() // Should not panic
 }
 
+func TestConversionQueueSerialBackgroundTasks(t *testing.T) {
+	queue := newConversionQueue()
+
+	firstStarted := make(chan struct{})
+	firstRelease := make(chan struct{})
+	if err := queue.launchConversion(func() {
+		close(firstStarted)
+		<-firstRelease
+	}); err != nil {
+		t.Fatalf("Failed to launch first conversion: %v", err)
+	}
+	<-firstStarted
+
+	secondStarted := make(chan struct{})
+	if err := queue.launchConversion(func() { close(secondStarted) }); err != nil {
+		t.Fatalf("Failed to launch second conversion: %v", err)
+	}
+	select {
+	case <-secondStarted:
+		close(firstRelease)
+		queue.close()
+		t.Fatal("Second conversion started before first conversion finished")
+	case <-time.After(100 * time.Millisecond):
+	}
+	close(firstRelease)
+	select {
+	case <-secondStarted:
+	case <-time.After(time.Second):
+		queue.close()
+		t.Fatal("Second conversion did not start after first conversion finished")
+	}
+	queue.close()
+}
+
+func TestConversionQueueCloseWaitsForBackgroundTask(t *testing.T) {
+	queue := newConversionQueue()
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	if err := queue.launchConversion(func() {
+		close(started)
+		<-release
+	}); err != nil {
+		t.Fatalf("Failed to launch conversion: %v", err)
+	}
+	<-started
+
+	closed := make(chan struct{})
+	go func() {
+		queue.close()
+		close(closed)
+	}()
+	select {
+	case <-closed:
+		t.Fatal("Queue closed before the running conversion finished")
+	case <-time.After(100 * time.Millisecond):
+	}
+	close(release)
+	select {
+	case <-closed:
+	case <-time.After(time.Second):
+		t.Fatal("Queue did not close after the running conversion finished")
+	}
+}
+
 func TestConversionQueueAutoRestartBatch(t *testing.T) {
 	queue := newConversionQueue()
 	defer queue.close()