handle channel closures in writer

xmtp · Jan 16, 2025 · cf48f8c · cf48f8c
1 parent 7f8e846
commit cf48f8c
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/pkg/blockchain/rpcLogStreamer.go b/pkg/blockchain/rpcLogStreamer.go
@@ -129,7 +129,7 @@ func (r *RpcLogStreamer) watchContract(watcher contractConfig) {
 	logger := r.logger.With(zap.String("contractAddress", watcher.contractAddress.Hex()))
 	startTime := time.Now()
 	defer close(watcher.eventChannel)
-	defer close(watcher.reorgChannel)
+
 	for {
 		select {
 		case <-r.ctx.Done():

diff --git a/pkg/indexer/indexer.go b/pkg/indexer/indexer.go
@@ -49,6 +49,12 @@ func NewIndexer(
 func (s *Indexer) Close() {
 	s.log.Debug("Closing")
 	if s.streamer != nil {
+		if s.streamer.messagesReorgChannel != nil {
+			close(s.streamer.messagesReorgChannel)
+		}
+		if s.streamer.identityUpdatesReorgChannel != nil {
+			close(s.streamer.identityUpdatesReorgChannel)
+		}
 		s.streamer.streamer.Stop()
 	}
 	s.cancel()