go.mod: update to go1.25, use waitgroup.Go, and enable modernize linter

.golangci.yml

@@ -9,6 +9,7 @@ linters:
   enable:
     - misspell
     - ineffassign
+    - modernize
     - revive
     - unconvert
     - unused

agent/worker.go

@@ -612,15 +612,13 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe
 			slices.Contains(sel.NodeIDs, t.NodeID)
 	}
 
-	wg := sync.WaitGroup{}
+	var wg sync.WaitGroup
 	w.mu.Lock()
 	for _, tm := range w.taskManagers {
 		if match(tm.task) {
-			wg.Add(1)
-			go func(tm *taskManager) {
-				defer wg.Done()
+			wg.Go(func() {
 				tm.Logs(ctx, *subscription.Options, publisher)
-			}(tm)
+			})
 		}
 	}
 	w.mu.Unlock()

go.mod

@@ -1,6 +1,6 @@
 module github.com/moby/swarmkit/v2
 
-go 1.24.0
+go 1.25.0
 
 require (
 	code.cloudfoundry.org/clock v1.1.0

integration/cluster_test.go

@@ -14,7 +14,6 @@ import (
 	"github.com/moby/swarmkit/v2/identity"
 	"github.com/moby/swarmkit/v2/log"
 	"github.com/moby/swarmkit/v2/manager/encryption"
-	"github.com/moby/swarmkit/v2/node"
 	"github.com/moby/swarmkit/v2/testutils"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -164,20 +163,17 @@ func (c *testCluster) runNode(n *testNode, nodeOrder int) error {
 	defer cancel()
 	defer close(done)
 
-	c.wg.Add(2)
-	go func() {
+	c.wg.Go(func() {
 		c.errs <- n.node.Start(ctx)
-		c.wg.Done()
-	}()
-	go func(n *node.Node) {
-		err := n.Err(errCtx)
+	})
+	c.wg.Go(func() {
+		err := n.node.Err(errCtx)
 		select {
 		case <-errCtx.Done():
 		default:
 			done <- err
 		}
-		c.wg.Done()
-	}(n.node)
+	})
 
 	select {
 	case <-n.node.Ready():

manager/allocator/allocator.go

@@ -106,28 +106,19 @@ func (a *Allocator) Run(ctx context.Context) error {
 			a.registerToVote(aa.taskVoter)
 		}
 
-		// Assign a pointer for variable capture
-		aaPtr := &aa
 		actor := func() error {
-			wg.Add(1)
-			defer wg.Done()
-
 			// init might return an allocator specific context
 			// which is a child of the passed in context to hold
 			// allocator specific state
-			watch, watchCancel, err := a.init(ctx, aaPtr)
+			watch, watchCancel, err := a.init(ctx, &aa)
 			if err != nil {
 				return err
 			}
 
-			wg.Add(1)
-			go func(watch <-chan events.Event, watchCancel func()) {
-				defer func() {
-					watchCancel()
-					wg.Done()
-				}()
-				a.run(ctx, *aaPtr, watch)
-			}(watch, watchCancel)
+			wg.Go(func() {
+				defer watchCancel()
+				a.run(ctx, aa, watch)
+			})
 			return nil
 		}
 

manager/controlapi/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"reflect"
+	"slices"
 	"strings"
 	"time"
 
@@ -247,10 +248,8 @@ func validateGenericRuntimeSpec(taskSpec api.TaskSpec) error {
 	}
 
 	reservedNames := []string{"container", "attachment"}
-	for _, n := range reservedNames {
-		if strings.ToLower(generic.Kind) == n {
-			return status.Errorf(codes.InvalidArgument, "Generic runtime: %q is a reserved name", generic.Kind)
-		}
+	if slices.Contains(reservedNames, strings.ToLower(generic.Kind)) {
+		return status.Errorf(codes.InvalidArgument, "Generic runtime: %q is a reserved name", generic.Kind)
 	}
 
 	payload := generic.Payload

manager/logbroker/broker_test.go

@@ -72,15 +72,12 @@ func TestLogBrokerLogs(t *testing.T) {
 			taskID := fmt.Sprintf("%v.task-%v", serviceID, task)
 
 			for node := range nNodes {
-				nodeID := fmt.Sprintf("node-%v", node)
-
 				if (task+1)%(node+1) != 0 {
 					continue
 				}
 				messagesExpected += nLogMessagesPerTask
 
-				wg.Add(1)
-				go func(nodeID, serviceID, taskID string) {
+				wg.Go(func() {
 					<-hold
 
 					// Each goroutine gets its own publisher
@@ -90,7 +87,6 @@ func TestLogBrokerLogs(t *testing.T) {
 					defer func() {
 						_, err := publisher.CloseAndRecv()
 						require.NoError(t, err)
-						wg.Done()
 					}()
 
 					msgctx := api.LogContext{
@@ -104,7 +100,7 @@ func TestLogBrokerLogs(t *testing.T) {
 							Messages:       []api.LogMessage{newLogMessage(msgctx, "log message number %d", i)},
 						}))
 					}
-				}(nodeID, serviceID, taskID)
+				})
 			}
 		}
 	}

manager/orchestrator/update/updater.go

@@ -189,13 +189,11 @@ func (u *Updater) Run(ctx context.Context, slots []orchestrator.Slot) {
 
 	// Start the workers.
 	slotQueue := make(chan orchestrator.Slot)
-	wg := sync.WaitGroup{}
-	wg.Add(parallelism)
+	var wg sync.WaitGroup
 	for range parallelism {
-		go func() {
+		wg.Go(func() {
 			u.worker(ctx, slotQueue, updateConfig)
-			wg.Done()
-		}()
+		})
 	}
 
 	var failedTaskWatch chan events.Event

manager/state/raft/raft.go

@@ -114,8 +114,8 @@ type Node struct {
 	reqIDGen            *idutil.Generator
 	wait                *wait
 	campaignWhenAble    bool
-	signalledLeadership uint32
-	isMember            uint32
+	signalledLeadership atomic.Uint32
+	isMember            atomic.Uint32
 	bootstrapMembers    []*api.RaftMember
 
 	// waitProp waits for all the proposals to be terminated before
@@ -161,7 +161,7 @@ type Node struct {
 	// an raft DEK during a raft DEK rotation, so that we won't finish a rotation until
 	// a snapshot covering that index has been written encrypted with the new raft DEK
 	waitForAppliedIndex uint64
-	ticksWithNoLeader   uint32
+	ticksWithNoLeader   atomic.Uint32
 }
 
 // NodeOptions provides node-level options.
@@ -275,7 +275,7 @@ func (n *Node) IsIDRemoved(id uint64) bool {
 // Part of transport.Raft interface.
 func (n *Node) NodeRemoved() {
 	n.removeRaftOnce.Do(func() {
-		atomic.StoreUint32(&n.isMember, 0)
+		n.isMember.Store(0)
 		close(n.RemovedFromRaft)
 	})
 }
@@ -331,7 +331,7 @@ func (n *Node) SetAddr(ctx context.Context, addr string) error {
 	ctx, cancelCtx := n.WithContext(ctx)
 	defer cancelCtx()
 
-	isLeader := atomic.LoadUint32(&n.signalledLeadership) == 1
+	isLeader := n.signalledLeadership.Load() == 1
 	for !isLeader {
 		select {
 		case leadershipChange := <-leadershipCh:
@@ -383,7 +383,7 @@ func (n *Node) JoinAndStart(ctx context.Context) (err error) {
 			n.stopMu.Unlock()
 			n.done()
 		} else {
-			atomic.StoreUint32(&n.isMember, 1)
+			n.isMember.Store(1)
 		}
 	}()
 
@@ -574,9 +574,9 @@ func (n *Node) Run(ctx context.Context) error {
 			n.raftNode.Tick()
 
 			if n.leader() == raft.None {
-				atomic.AddUint32(&n.ticksWithNoLeader, 1)
+				n.ticksWithNoLeader.Add(1)
 			} else {
-				atomic.StoreUint32(&n.ticksWithNoLeader, 0)
+				n.ticksWithNoLeader.Store(0)
 			}
 		case rd := <-n.raftNode.Ready():
 			raftConfig := n.getCurrentRaftConfig()
@@ -646,8 +646,8 @@ func (n *Node) Run(ctx context.Context) error {
 					wasLeader = false
 					log.G(ctx).Error("soft state changed, node no longer a leader, resetting and cancelling all waits")
 
-					if atomic.LoadUint32(&n.signalledLeadership) == 1 {
-						atomic.StoreUint32(&n.signalledLeadership, 0)
+					if n.signalledLeadership.Load() == 1 {
+						n.signalledLeadership.Store(0)
 						n.leadershipBroadcast.Publish(IsFollower)
 					}
 
@@ -686,11 +686,11 @@ func (n *Node) Run(ctx context.Context) error {
 				n.triggerSnapshot(ctx, raftConfig)
 			}
 
-			if wasLeader && atomic.LoadUint32(&n.signalledLeadership) != 1 {
+			if wasLeader && n.signalledLeadership.Load() != 1 {
 				// If all the entries in the log have become
 				// committed, broadcast our leadership status.
 				if n.caughtUp() {
-					atomic.StoreUint32(&n.signalledLeadership, 1)
+					n.signalledLeadership.Store(1)
 					n.leadershipBroadcast.Publish(IsLeader)
 				}
 			}
@@ -860,7 +860,7 @@ func (n *Node) stop(ctx context.Context) {
 	n.raftNode.Stop()
 	n.ticker.Stop()
 	n.raftLogger.Close(ctx)
-	atomic.StoreUint32(&n.isMember, 0)
+	n.isMember.Store(0)
 	// TODO(stevvooe): Handle ctx.Done()
 }
 
@@ -910,7 +910,7 @@ func (n *Node) Leader() (uint64, error) {
 // saying that it has become the leader. This means it is ready to accept
 // proposals.
 func (n *Node) ReadyForProposals() bool {
-	return atomic.LoadUint32(&n.signalledLeadership) == 1
+	return n.signalledLeadership.Load() == 1
 }
 
 func (n *Node) caughtUp() bool {
@@ -1517,7 +1517,7 @@ func (n *Node) LeaderConn(ctx context.Context) (*grpc.ClientConn, error) {
 	if err == raftselector.ErrIsLeader {
 		return nil, err
 	}
-	if atomic.LoadUint32(&n.ticksWithNoLeader) > lostQuorumTimeout {
+	if n.ticksWithNoLeader.Load() > lostQuorumTimeout {
 		return nil, errLostQuorum
 	}
 
@@ -1732,7 +1732,7 @@ func (n *Node) GetNodeIDByRaftID(raftID uint64) (string, error) {
 // IsMember checks if the raft node has effectively joined
 // a cluster of existing members.
 func (n *Node) IsMember() bool {
-	return atomic.LoadUint32(&n.isMember) == 1
+	return n.isMember.Load() == 1
 }
 
 // Saves a log entry to our Store
@@ -1801,7 +1801,7 @@ func (n *Node) processInternalRaftRequest(ctx context.Context, r *api.InternalRa
 	ch := n.wait.register(r.ID, cb, cancel)
 
 	// Do this check after calling register to avoid a race.
-	if atomic.LoadUint32(&n.signalledLeadership) != 1 {
+	if n.signalledLeadership.Load() != 1 {
 		log.G(ctx).Error("node is no longer leader, aborting propose")
 		n.wait.cancel(r.ID)
 		return nil, ErrLostLeadership
@@ -1829,7 +1829,7 @@ func (n *Node) processInternalRaftRequest(ctx context.Context, r *api.InternalRa
 		if !ok {
 			// Wait notification channel was closed. This should only happen if the wait was cancelled.
 			log.G(ctx).Error("wait cancelled")
-			if atomic.LoadUint32(&n.signalledLeadership) == 1 {
+			if n.signalledLeadership.Load() == 1 {
 				log.G(ctx).Error("wait cancelled but node is still a leader")
 			}
 			return nil, ErrLostLeadership
@@ -1841,7 +1841,7 @@ func (n *Node) processInternalRaftRequest(ctx context.Context, r *api.InternalRa
 		x, ok := <-ch
 		if !ok {
 			log.G(ctx).WithError(waitCtx.Err()).Error("wait context cancelled")
-			if atomic.LoadUint32(&n.signalledLeadership) == 1 {
+			if n.signalledLeadership.Load() == 1 {
 				log.G(ctx).Error("wait context cancelled but node is still a leader")
 			}
 			return nil, ErrLostLeadership

manager/state/store/memory_test.go

@@ -2115,9 +2115,7 @@ func BenchmarkNodeConcurrency(b *testing.B) {
 	// Run 5 writer goroutines and 5 reader goroutines
 	var wg sync.WaitGroup
 	for c := range 5 {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			for i := range b.N {
 				_ = s.Update(func(tx1 Tx) error {
 					_ = UpdateNode(tx1, &api.Node{
@@ -2131,19 +2129,17 @@ func BenchmarkNodeConcurrency(b *testing.B) {
 					return nil
 				})
 			}
-		}()
+		})
 	}
 
 	for range 5 {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			s.View(func(tx1 ReadTx) {
 				for i := range b.N {
 					_ = GetNode(tx1, nodeIDs[i%benchmarkNumNodes])
 				}
 			})
-		}()
+		})
 	}
 
 	wg.Wait()