Merge pull request #53208 from renovate-bot/renovate/github.com-moby-swarmkit-v2-digest

fix(deps): update github.com/moby/swarmkit/v2 digest to e53c2e1

go.mod

@@ -74,7 +74,7 @@ require (
 	github.com/moby/profiles/apparmor v0.2.1
 	github.com/moby/profiles/seccomp v0.2.3
 	github.com/moby/pubsub v1.0.0
-	github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35
+	github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9
 	github.com/moby/sys/atomicwriter v0.1.0
 	github.com/moby/sys/mount v0.3.5
 	github.com/moby/sys/mountinfo v0.7.2

go.sum

@@ -563,8 +563,8 @@ github.com/moby/profiles/seccomp v0.2.3 h1:nrHNSiECQQvq4WjgceCUgJXIXUJBIswVQ133k
 github.com/moby/profiles/seccomp v0.2.3/go.mod h1:8m3qkkWZXrRsMqlUUN2zyccnYmBf3EAdQYPMVJ3NBhk=
 github.com/moby/pubsub v1.0.0 h1:jkp/imWsmJz2f6LyFsk7EkVeN2HxR/HTTOY8kHrsxfA=
 github.com/moby/pubsub v1.0.0/go.mod h1:bXSO+3h5MNXXCaEG+6/NlAIk7MMZbySZlnB+cUQhKKc=
-github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35 h1:8D44vdVynl7eWS5XblHfyctMHcrzbLS4YSVcZBvjwKY=
-github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35/go.mod h1:9rg8/RJeagBT8uRrsNFcetujUVZT/ynfya2hnabQY/A=
+github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9 h1:uv2uVSLLXQtAZOK9s4vXyBy7vOZHGKXEUf+qCp0KSCs=
+github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9/go.mod h1:9rg8/RJeagBT8uRrsNFcetujUVZT/ynfya2hnabQY/A=
 github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w3A14Sw=
 github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs=
 github.com/moby/sys/capability v0.4.0 h1:4D4mI6KlNtWMCM1Z/K0i7RV1FkX+DBDHKVJpCndZoHk=

vendor/github.com/moby/swarmkit/v2/manager/logbroker/broker.go

@@ -105,40 +105,38 @@ func (lb *LogBroker) newSubscription(selector *api.LogSelector, options *api.Log
 	lb.mu.RLock()
 	defer lb.mu.RUnlock()
 
-	subscription := newSubscription(lb.store, &api.SubscriptionMessage{
+	return newSubscription(lb.store, &api.SubscriptionMessage{
 		ID:       identity.NewID(),
 		Selector: selector,
 		Options:  options,
 	}, lb.subscriptionQueue)
-
-	return subscription
 }
 
 func (lb *LogBroker) getSubscription(id string) *subscription {
 	lb.mu.RLock()
 	defer lb.mu.RUnlock()
 
-	subscription, ok := lb.registeredSubscriptions[id]
+	sub, ok := lb.registeredSubscriptions[id]
 	if !ok {
 		return nil
 	}
-	return subscription
+	return sub
 }
 
-func (lb *LogBroker) registerSubscription(subscription *subscription) {
+func (lb *LogBroker) registerSubscription(sub *subscription) {
 	lb.mu.Lock()
 	defer lb.mu.Unlock()
 
-	lb.registeredSubscriptions[subscription.message.ID] = subscription
-	lb.subscriptionQueue.Publish(subscription)
+	lb.registeredSubscriptions[sub.ID()] = sub
+	lb.subscriptionQueue.Publish(sub)
 
-	for _, node := range subscription.Nodes() {
+	for _, node := range sub.Nodes() {
 		if _, ok := lb.subscriptionsByNode[node]; !ok {
 			// Mark nodes that won't receive the message as done.
-			subscription.Done(node, fmt.Errorf("node %s is not available", node))
+			sub.Done(node, fmt.Errorf("node %s is not available", node))
 		} else {
 			// otherwise, add the subscription to the node's subscriptions list
-			lb.subscriptionsByNode[node][subscription] = struct{}{}
+			lb.subscriptionsByNode[node][sub] = struct{}{}
 		}
 	}
 }
@@ -147,7 +145,7 @@ func (lb *LogBroker) unregisterSubscription(subscription *subscription) {
 	lb.mu.Lock()
 	defer lb.mu.Unlock()
 
-	delete(lb.registeredSubscriptions, subscription.message.ID)
+	delete(lb.registeredSubscriptions, subscription.ID())
 
 	// remove the subscription from all of the nodes
 	for _, node := range subscription.Nodes() {
@@ -234,25 +232,25 @@ func (lb *LogBroker) SubscribeLogs(request *api.SubscribeLogsRequest, stream api
 		return errNotRunning
 	}
 
-	subscription := lb.newSubscription(request.Selector, request.Options)
-	subscription.Run(pctx)
-	defer subscription.Stop()
+	sub := lb.newSubscription(request.Selector, request.Options)
+	sub.Run(pctx)
+	defer sub.Stop()
 
 	logger := log.G(ctx).WithFields(
 		log.Fields{
 			"method":          "(*LogBroker).SubscribeLogs",
-			"subscription.id": subscription.message.ID,
+			"subscription.id": sub.ID(),
 		},
 	)
 	logger.Debug("subscribed")
 
-	publishCh, publishCancel := lb.subscribe(subscription.message.ID)
+	publishCh, publishCancel := lb.subscribe(sub.ID())
 	defer publishCancel()
 
-	lb.registerSubscription(subscription)
-	defer lb.unregisterSubscription(subscription)
+	lb.registerSubscription(sub)
+	defer lb.unregisterSubscription(sub)
 
-	completed := subscription.Wait(ctx)
+	completed := sub.Wait(ctx)
 	for {
 		select {
 		case <-ctx.Done():
@@ -273,10 +271,10 @@ func (lb *LogBroker) SubscribeLogs(request *api.SubscribeLogsRequest, stream api
 			completed = nil
 			lb.logQueue.Publish(&logMessage{
 				PublishLogsMessage: &api.PublishLogsMessage{
-					SubscriptionID: subscription.message.ID,
+					SubscriptionID: sub.ID(),
 				},
 				completed: true,
-				err:       subscription.Err(),
+				err:       sub.Err(),
 			})
 		}
 	}
@@ -332,7 +330,7 @@ func (lb *LogBroker) ListenSubscriptions(_ *api.ListenSubscriptionsRequest, stre
 	activeSubscriptions := make(map[string]*subscription)
 
 	// Start by sending down all active subscriptions.
-	for _, subscription := range subscriptions {
+	for _, sub := range subscriptions {
 		select {
 		case <-stream.Context().Done():
 			return stream.Context().Err()
@@ -341,30 +339,30 @@ func (lb *LogBroker) ListenSubscriptions(_ *api.ListenSubscriptionsRequest, stre
 		default:
 		}
 
-		if err := stream.Send(subscription.message); err != nil {
-			logger.Error(err)
+		if err := stream.Send(sub.Message()); err != nil {
+			logger.WithError(err).Error("failed to send initial subscription")
 			return err
 		}
-		activeSubscriptions[subscription.message.ID] = subscription
+		activeSubscriptions[sub.ID()] = sub
 	}
 
 	// Send down new subscriptions.
 	for {
 		select {
 		case v := <-subscriptionCh:
-			subscription := v.(*subscription)
+			sub := v.(*subscription)
 
-			if subscription.Closed() {
-				delete(activeSubscriptions, subscription.message.ID)
+			if sub.Closed() {
+				delete(activeSubscriptions, sub.ID())
 			} else {
 				// Avoid sending down the same subscription multiple times
-				if _, ok := activeSubscriptions[subscription.message.ID]; ok {
+				if _, ok := activeSubscriptions[sub.ID()]; ok {
 					continue
 				}
-				activeSubscriptions[subscription.message.ID] = subscription
+				activeSubscriptions[sub.ID()] = sub
 			}
-			if err := stream.Send(subscription.message); err != nil {
-				logger.Error(err)
+			if err := stream.Send(sub.Message()); err != nil {
+				logger.WithError(err).Error("failed to send subscription update")
 				return err
 			}
 		case <-stream.Context().Done():
@@ -408,7 +406,7 @@ func (lb *LogBroker) PublishLogs(stream api.LogBroker_PublishLogsServer) (err er
 				return status.Errorf(codes.NotFound, "unknown subscription ID")
 			}
 		} else {
-			if logMsg.SubscriptionID != currentSubscription.message.ID {
+			if logMsg.SubscriptionID != currentSubscription.ID() {
 				return status.Errorf(codes.InvalidArgument, "different subscription IDs in the same session")
 			}
 		}

vendor/github.com/moby/swarmkit/v2/manager/logbroker/subscription.go

@@ -15,6 +15,10 @@ import (
 )
 
 type subscription struct {
+	id string
+
+	follow bool
+
 	mu sync.RWMutex
 	wg sync.WaitGroup
 
@@ -32,6 +36,8 @@ type subscription struct {
 
 func newSubscription(store *store.MemoryStore, message *api.SubscriptionMessage, changed *watch.Queue) *subscription {
 	return &subscription{
+		id:           message.ID,
+		follow:       message.Options != nil && message.Options.Follow,
 		store:        store,
 		message:      message,
 		changed:      changed,
@@ -40,8 +46,8 @@ func newSubscription(store *store.MemoryStore, message *api.SubscriptionMessage,
 	}
 }
 
-func (s *subscription) follow() bool {
-	return s.message.Options != nil && s.message.Options.Follow
+func (s *subscription) ID() string {
+	return s.id
 }
 
 func (s *subscription) Contains(nodeID string) bool {
@@ -66,12 +72,12 @@ func (s *subscription) Nodes() []string {
 func (s *subscription) Run(ctx context.Context) {
 	s.ctx, s.cancel = context.WithCancel(ctx)
 
-	if s.follow() {
+	if s.follow {
 		wq := s.store.WatchQueue()
 		ch, cancel := state.Watch(wq, api.EventCreateTask{}, api.EventUpdateTask{})
 		go func() {
 			defer cancel()
-			s.watch(ch)
+			_ = s.watch(ch)
 		}()
 	}
 
@@ -86,7 +92,7 @@ func (s *subscription) Stop() {
 
 func (s *subscription) Wait(_ context.Context) <-chan struct{} {
 	// Follow subscriptions never end
-	if s.follow() {
+	if s.follow {
 		return nil
 	}
 
@@ -106,7 +112,7 @@ func (s *subscription) Done(nodeID string, err error) {
 		s.errors = append(s.errors, err)
 	}
 
-	if s.follow() {
+	if s.follow {
 		return
 	}
 
@@ -152,6 +158,7 @@ func (s *subscription) Closed() bool {
 func (s *subscription) match() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	selector := s.message.Selector
 
 	add := func(t *api.Task) {
 		if t.NodeID == "" {
@@ -165,17 +172,17 @@ func (s *subscription) match() {
 	}
 
 	s.store.View(func(tx store.ReadTx) {
-		for _, nid := range s.message.Selector.NodeIDs {
+		for _, nid := range selector.NodeIDs {
 			s.nodes[nid] = struct{}{}
 		}
 
-		for _, tid := range s.message.Selector.TaskIDs {
+		for _, tid := range selector.TaskIDs {
 			if task := store.GetTask(tx, tid); task != nil {
 				add(task)
 			}
 		}
 
-		for _, sid := range s.message.Selector.ServiceIDs {
+		for _, sid := range selector.ServiceIDs {
 			tasks, err := store.FindTasks(tx, store.ByServiceID(sid))
 			if err != nil {
 				log.L.Warning(err)
@@ -183,7 +190,7 @@ func (s *subscription) match() {
 			}
 			for _, task := range tasks {
 				// if we're not following, don't add tasks that aren't running yet
-				if !s.follow() && task.Status.State < api.TaskStateRunning {
+				if !s.follow && task.Status.State < api.TaskStateRunning {
 					continue
 				}
 				add(task)
@@ -193,13 +200,15 @@ func (s *subscription) match() {
 }
 
 func (s *subscription) watch(ch <-chan events.Event) error {
+	selector := s.message.Selector
+
 	matchTasks := map[string]struct{}{}
-	for _, tid := range s.message.Selector.TaskIDs {
+	for _, tid := range selector.TaskIDs {
 		matchTasks[tid] = struct{}{}
 	}
 
 	matchServices := map[string]struct{}{}
-	for _, sid := range s.message.Selector.ServiceIDs {
+	for _, sid := range selector.ServiceIDs {
 		matchServices[sid] = struct{}{}
 	}
 
@@ -257,3 +266,18 @@ func (s *subscription) watch(ch <-chan events.Event) error {
 		}
 	}
 }
+
+// Message returns a snapshot of the current subscription message.
+//
+// The underlying message is owned by the subscription and may be updated
+// as part of its lifecycle. Callers should use this accessor rather than
+// accessing the underlying message directly.
+func (s *subscription) Message() *api.SubscriptionMessage {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	// Return a snapshot of the message to avoid races while it is being
+	// marshaled. See https://github.com/moby/moby/issues/47322.
+	msg := *s.message
+	return &msg
+}

vendor/modules.txt

@@ -1331,7 +1331,7 @@ github.com/moby/profiles/seccomp
 # github.com/moby/pubsub v1.0.0
 ## explicit; go 1.19
 github.com/moby/pubsub
-# github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35
+# github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9
 ## explicit; go 1.24.0
 github.com/moby/swarmkit/v2/agent
 github.com/moby/swarmkit/v2/agent/configs