handle channel closures in writer

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

Hide file tree

Showing 2 changed files with 14 additions and 8 deletions.
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
@@ -46,14 +46,20 @@ func NewIndexer(
 	}
 }
 
-func (s *Indexer) Close() {
-	s.log.Debug("Closing")
-	if s.streamer != nil {
-		s.streamer.streamer.Stop()
+func (i *Indexer) Close() {
+	i.log.Debug("Closing")
+	if i.streamer != nil {
+		if i.streamer.messagesReorgChannel != nil {
+			close(i.streamer.messagesReorgChannel)
+		}
+		if i.streamer.identityUpdatesReorgChannel != nil {
+			close(i.streamer.identityUpdatesReorgChannel)
+		}
+		i.streamer.streamer.Stop()
 	}
-	s.cancel()
-	s.wg.Wait()
-	s.log.Debug("Closed")
+	i.cancel()
+	i.wg.Wait()
+	i.log.Debug("Closed")
 }
 
 func (i *Indexer) StartIndexer(