fix stream api deadlock

amplitude · Oct 21, 2024 · 72d1fd1 · 72d1fd1
1 parent a896a48
commit 72d1fd1
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 7 deletions.
diff --git a/pkg/experiment/local/flag_config_stream_api.go b/pkg/experiment/local/flag_config_stream_api.go
@@ -119,17 +119,14 @@ func (api *flagConfigStreamApiV2) Connect(
 	stopCh := make(chan bool)
 	api.stopCh = stopCh
 
-	closeAllAndNotify := func(err error) {
+	closeAll := func() {
 		api.lock.Lock()
 		defer api.lock.Unlock()
 		closeStream()
 		if api.stopCh == stopCh {
 			api.stopCh = nil
 		}
 		close(stopCh)
-		if onError != nil {
-			onError(err)
-		}
 	}
 
 	// Retrieve and pass on message forever until stopCh closes.
@@ -144,7 +141,10 @@ func (api *flagConfigStreamApiV2) Connect(
 				flags, err := parseData(msg.data)
 				if err != nil {
 					// Error, close everything.
-					closeAllAndNotify(errors.New("stream corrupt data, cause: " + err.Error()))
+					closeAll()
+					if onError != nil {
+						onError(errors.New("stream corrupt data, cause: " + err.Error()))
+					}
 					return
 				}
 				if onUpdate != nil {
@@ -154,7 +154,10 @@ func (api *flagConfigStreamApiV2) Connect(
 				}
 			case err := <-streamErrCh:
 				// Error, close everything.
-				closeAllAndNotify(err)
+				closeAll()
+				if onError != nil {
+					onError(err)
+				}
 				return
 			}
 		}

diff --git a/pkg/experiment/local/flag_config_updater.go b/pkg/experiment/local/flag_config_updater.go
@@ -354,7 +354,8 @@ func (w *flagConfigFallbackRetryWrapper) scheduleRetry() {
 		}
 
 		w.log.Debug("main updater retry start")
-		err := w.mainUpdater.Start(func(error) {
+		err := w.mainUpdater.Start(func(err error) {
+			w.log.Error("main updater updating err, starting fallback if available. error: ", err)
 			go func() { w.scheduleRetry() }() // Don't care if poller start error or not, always retry.
 			if w.fallbackUpdater != nil {
 				//nolint:errcheck