fix: add runJob helper for non-blocking semaphore pattern

Author: minhd-vu

p2p/database/datastore.go

@@ -129,6 +129,20 @@ type DatastoreOptions struct {
 	TTL                          time.Duration
 }
 
+// runJob executes fn in a goroutine with semaphore-based concurrency control.
+// It will wait for a semaphore slot without blocking the caller, and respects context cancellation.
+func (d *Datastore) runJob(ctx context.Context, fn func()) {
+	go func() {
+		select {
+		case d.jobs <- struct{}{}:
+			fn()
+			<-d.jobs
+		case <-ctx.Done():
+			return
+		}
+	}()
+}
+
 // NewDatastore connects to datastore and creates the client. This should
 // only be called once unless trying to write to different databases.
 func NewDatastore(ctx context.Context, opts DatastoreOptions) Database {
@@ -159,19 +173,15 @@ func (d *Datastore) WriteBlock(ctx context.Context, peer *enode.Node, block *typ
 	}
 
 	if d.ShouldWriteBlockEvents() {
-		d.jobs <- struct{}{}
-		go func() {
+		d.runJob(ctx, func() {
 			d.writeEvent(peer, BlockEventsKind, block.Hash(), BlocksKind, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 
 	if d.ShouldWriteBlocks() {
-		d.jobs <- struct{}{}
-		go func() {
+		d.runJob(ctx, func() {
 			d.writeBlock(ctx, block, td, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 }
 
@@ -185,11 +195,10 @@ func (d *Datastore) WriteBlockHeaders(ctx context.Context, headers []*types.Head
 	}
 
 	for _, h := range headers {
-		d.jobs <- struct{}{}
-		go func(header *types.Header) {
+		header := h
+		d.runJob(ctx, func() {
 			d.writeBlockHeader(ctx, header, tfs)
-			<-d.jobs
-		}(h)
+		})
 	}
 }
 
@@ -203,11 +212,9 @@ func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, has
 		return
 	}
 
-	d.jobs <- struct{}{}
-	go func() {
+	d.runJob(ctx, func() {
 		d.writeBlockBody(ctx, body, hash, tfs)
-		<-d.jobs
-	}()
+	})
 }
 
 // WriteBlockHashes will write the block events to datastore.
@@ -216,11 +223,9 @@ func (d *Datastore) WriteBlockHashes(ctx context.Context, peer *enode.Node, hash
 		return
 	}
 
-	d.jobs <- struct{}{}
-	go func() {
+	d.runJob(ctx, func() {
 		d.writeEvents(ctx, peer, BlockEventsKind, hashes, BlocksKind, tfs)
-		<-d.jobs
-	}()
+	})
 }
 
 // WriteTransactions will write the transactions and transaction events to datastore.
@@ -230,11 +235,9 @@ func (d *Datastore) WriteTransactions(ctx context.Context, peer *enode.Node, txs
 	}
 
 	if d.ShouldWriteTransactions() {
-		d.jobs <- struct{}{}
-		go func() {
+		d.runJob(ctx, func() {
 			d.writeTransactions(ctx, txs, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 
 	if d.ShouldWriteTransactionEvents() {
@@ -243,11 +246,9 @@ func (d *Datastore) WriteTransactions(ctx context.Context, peer *enode.Node, txs
 			hashes = append(hashes, tx.Hash())
 		}
 
-		d.jobs <- struct{}{}
-		go func() {
+		d.runJob(ctx, func() {
 			d.writeEvents(ctx, peer, TransactionEventsKind, hashes, TransactionsKind, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 }
 
@@ -257,9 +258,7 @@ func (d *Datastore) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.
 		return
 	}
 
-	d.jobs <- struct{}{}
-	go func() {
-
+	d.runJob(ctx, func() {
 		keys := make([]*datastore.Key, 0, len(peers))
 		dsPeers := make([]*DatastorePeer, 0, len(peers))
 
@@ -279,9 +278,7 @@ func (d *Datastore) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.
 		if err != nil {
 			log.Error().Err(err).Msg("Failed to write peers")
 		}
-
-		<-d.jobs
-	}()
+	})
 }
 
 func (d *Datastore) MaxConcurrentWrites() int {