fix pubsub race

PeerDB-io · May 7, 2024 · ec35652 · ec35652
1 parent 1f084e7
commit ec35652
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 3 deletions.
diff --git a/flow/connectors/pubsub/pubsub.go b/flow/connectors/pubsub/pubsub.go
@@ -325,10 +325,10 @@ Loop:
 	}
 
 	close(flushLoopDone)
-	close(publish)
 	if err := pool.Wait(wgCtx); err != nil {
 		return nil, err
 	}
+	close(publish)
 	topiccache.Stop(wgCtx)
 	select {
 	case <-wgCtx.Done():

diff --git a/flow/connectors/pubsub/qrep.go b/flow/connectors/pubsub/qrep.go
@@ -110,10 +110,10 @@ Loop:
 		}
 	}
 
-	close(publish)
 	if err := pool.Wait(wgCtx); err != nil {
 		return 0, err
 	}
+	close(publish)
 	topiccache.Stop(wgCtx)
 	select {
 	case <-wgCtx.Done():

diff --git a/flow/e2e/pubsub/pubsub_test.go b/flow/e2e/pubsub/pubsub_test.go
@@ -278,7 +278,6 @@ func (s PubSubSuite) TestInitialLoad() {
 	})
 	require.NoError(s.t, err)
 	_, err = s.Conn().Exec(context.Background(), fmt.Sprintf(`
-
 		INSERT INTO %s (id, val) VALUES (1, 'testval')
 	`, srcTableName))
 	require.NoError(s.t, err)