Merge pull request #40432 from thaJeztah/19.03_bump_swarmkit

[19.03] vendor: bump swarmkit 062b694b46c0744d601eebef79f3f7433d808a04

vendor.conf

@@ -128,7 +128,7 @@ github.com/containerd/ttrpc                         92c8520ef9f86600c650dd540266
 github.com/gogo/googleapis                          d31c731455cb061f42baff3bda55bad0118b126b # v1.2.0
 
 # cluster
-github.com/docker/swarmkit                          f35d9100f2c6ac810cc8d7de6e8f93dcc7a42d29 # bump_v19.03 branch
+github.com/docker/swarmkit                          062b694b46c0744d601eebef79f3f7433d808a04 # bump_v19.03 branch
 github.com/gogo/protobuf                            ba06b47c162d49f2af050fb4c75bcbc86a159d5c # v1.2.1
 github.com/golang/protobuf                          aa810b61a9c79d51363740d207bb46cf8e620ed5 # v1.2.0
 github.com/cloudflare/cfssl                         5d63dbd981b5c408effbb58c442d54761ff94fbd # 1.3.2

vendor/github.com/docker/swarmkit/agent/agent.go

@@ -205,6 +205,11 @@ func (a *Agent) run(ctx context.Context) {
 		sessionq      chan sessionOperation
 		leaving       = a.leaving
 		subscriptions = map[string]context.CancelFunc{}
+		// subscriptionDone is a channel that allows us to notify ourselves
+		// that a lot subscription should be finished. this channel is
+		// unbuffered, because it is only written to in a goroutine, and
+		// therefore cannot block the main execution path.
+		subscriptionDone = make(chan string)
 	)
 	defer func() {
 		session.close()
@@ -310,8 +315,26 @@ func (a *Agent) run(ctx context.Context) {
 
 			subCtx, subCancel := context.WithCancel(ctx)
 			subscriptions[sub.ID] = subCancel
-			// TODO(dperny) we're tossing the error here, that seems wrong
-			go a.worker.Subscribe(subCtx, sub)
+			// NOTE(dperny): for like 3 years, there has been a to do saying
+			// "we're tossing the error here, that seems wrong". this is not a
+			// to do anymore. 9/10 of these errors are going to be "context
+			// deadline exceeded", and the remaining 1/10 obviously doesn't
+			// matter or we'd have missed it by now.
+			go func() {
+				a.worker.Subscribe(subCtx, sub)
+				// when the worker finishes the subscription, we should notify
+				// ourselves that this has occurred. We cannot rely on getting
+				// a Close message from the manager, as any number of things
+				// could go wrong (see github.com/moby/moby/issues/39916).
+				subscriptionDone <- sub.ID
+			}()
+		case subID := <-subscriptionDone:
+			// subscription may already have been removed. If so, no need to
+			// take any action.
+			if cancel, ok := subscriptions[subID]; ok {
+				cancel()
+				delete(subscriptions, subID)
+			}
 		case <-registered:
 			log.G(ctx).Debugln("agent: registered")
 			if ready != nil {
@@ -548,8 +571,9 @@ func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogP
 			SubscriptionID: subscriptionID,
 			Close:          true,
 		})
-		// close the stream forreal
-		publisher.CloseSend()
+		// close the stream forreal. ignore the return value and the error,
+		// because we don't care.
+		publisher.CloseAndRecv()
 	}
 
 	return exec.LogPublisherFunc(func(ctx context.Context, message api.LogMessage) error {