fix(stream): wrap close done ch with sync.Once (aws#2537)

efekarakus · web-flow · commit 3d9721285d30 · 2021-06-28T23:08:10.000Z
Today, `stream.Stream` is structured as:
```go
case &lt;-streamer.Done():
  // call Close()
case &lt;-time.After(fetchDelay):
  // call Fetch()
```
The problem is that it's possible that the `Done()` event is triggered at
the same time as `time.After(fetchDelay)` and in that scenario the
behavior is undeterministic.
It's possible that we call `Fetch` a second time. For the ECS deployment
streamer, this means that we would previous close the `s.done` channel
twice. Wrapping the call with a `sync.Once` removes this concern.


By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.
diff --git a/internal/pkg/stream/ecs.go b/internal/pkg/stream/ecs.go
@@ -74,6 +74,7 @@ type ECSDeploymentStreamer struct {
 	deploymentCreationTime time.Time
 
 	subscribers   []chan ECSService
+	once          sync.Once
 	done          chan struct{}
 	isDone        bool
 	pastEventIDs  map[string]bool
@@ -143,7 +144,12 @@ func (s *ECSDeploymentStreamer) Fetch() (next time.Time, err error) {
 		deployments = append(deployments, rollingDeploy)
 		if isDeploymentDone(rollingDeploy, s.deploymentCreationTime) {
 			// The deployment is done, notify that there is no need for another Fetch call beyond this point.
-			close(s.done)
+			// In stream.Stream, it's possible that both the <-Done() event is available as well as another Fetch()
+			// call. In order to guarantee that we don't try to close the same stream multiple times, we wrap it with a
+			// sync.Once.
+			s.once.Do(func() {
+				close(s.done)
+			})
 		}
 	}
 	var failureMsgs []string
