From f9f90a9108aa75ac33d3323c73ef000e5fe7d63d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Louis=20G=C3=A9rard?= Date: Wed, 22 Jan 2025 18:52:51 +0100 Subject: [PATCH] fix(server): better logic on connection lsot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Louis Gérard --- internal/firehose/stream.go | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/internal/firehose/stream.go b/internal/firehose/stream.go index fd13a0e..9cfda74 100644 --- a/internal/firehose/stream.go +++ b/internal/firehose/stream.go @@ -95,11 +95,7 @@ func (fh *FirehoseListener) onPostCreate(ctx context.Context, post *bsky.FeedPos func (fh *FirehoseListener) SubscribeFirehose() { uri := "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos" - con, _, err := websocket.DefaultDialer.Dial(uri, http.Header{}) - if err != nil { - panic(err) - } - ctx := context.TODO() + ctx := context.Background() rsc := &events.RepoStreamCallbacks{ RepoCommit: func(evt *atproto.SyncSubscribeRepos_Commit) error { if evt.TooBig { @@ -143,13 +139,21 @@ func (fh *FirehoseListener) SubscribeFirehose() { }, } scheduler := sequential.NewScheduler("stream", rsc.EventHandler) - for { + maxRetries := 5 + delay := 10 * time.Second + currentRetry := 0 + for currentRetry < maxRetries { + fh.logger.Info("trying to connect", "try", currentRetry, "maxRetries", maxRetries) + con, _, err := websocket.DefaultDialer.Dial(uri, http.Header{}) + if err != nil { + fh.logger.Info("failed to establish connection", "err", err.Error()) + time.Sleep(delay * time.Duration(currentRetry)) + continue + } err = events.HandleRepoStream(ctx, con, scheduler, fh.logger) if err != nil { - delay := 10 * time.Second fh.logger.Error("handling repo stream", "err", err.Error()) - time.Sleep(delay) - fh.logger.Info("trying to reconnect", "", err.Error()) + time.Sleep(delay * time.Duration(currentRetry)) } } } -- 2.54.0