Low-latency AT Protocol interaction indexer.
firehose: reconnect with backoff
Aly Raffauf
(Jul 3, 2026, 1:37 PM EDT)
45df4942
05b1f39b
+37
-11
+2
-9
cmd/asterism/main.go
+2
-9
cmd/asterism/main.go
···
5
5
"flag"
6
6
"fmt"
7
7
"log/slog"
8
-
"net/http"
9
8
"strings"
10
9
11
10
"github.com/bluesky-social/indigo/atproto/identity"
12
11
"github.com/bluesky-social/indigo/xrpc"
13
-
"github.com/gorilla/websocket"
14
12
15
13
"github.com/alyraffauf/asterism/internal/api"
16
14
"github.com/alyraffauf/asterism/internal/backfill"
···
44
42
ctx := context.Background()
45
43
logger := slog.Default()
46
44
47
-
conn, _, err := websocket.DefaultDialer.Dial(relayURL, http.Header{})
48
-
if err != nil {
49
-
panic(err)
50
-
}
51
-
defer conn.Close()
52
-
53
45
linkStore, err := store.Open("asterism.db")
54
46
if err != nil {
55
47
panic(err)
···
89
81
Backfill: bf,
90
82
}
91
83
92
-
if err := consumer.Run(ctx, conn, logger); err != nil {
84
+
if err := consumer.Run(ctx, relayURL, logger); err != nil {
93
85
panic(err)
94
86
}
87
+
95
88
}
+35
-2
internal/firehose/stream.go
+35
-2
internal/firehose/stream.go
···
3
3
import (
4
4
"context"
5
5
"log/slog"
6
+
"net/http"
7
+
"time"
6
8
7
9
"github.com/bluesky-social/indigo/api/atproto"
8
10
"github.com/bluesky-social/indigo/events"
···
10
12
"github.com/gorilla/websocket"
11
13
)
12
14
13
-
func (c *Consumer) Run(ctx context.Context, conn *websocket.Conn, logger *slog.Logger) error {
15
+
const (
16
+
minBackoff = 1 * time.Second
17
+
maxBackoff = 30 * time.Second
18
+
)
19
+
20
+
func (c *Consumer) Run(ctx context.Context, relayURL string, logger *slog.Logger) error {
21
+
backoff := minBackoff
22
+
23
+
for {
24
+
conn, _, err := websocket.DefaultDialer.DialContext(ctx, relayURL, http.Header{})
25
+
if err != nil {
26
+
logger.Warn("dial failed", "err", err, "retry in", backoff)
27
+
} else {
28
+
backoff = minBackoff
29
+
c.stream(ctx, conn, logger)
30
+
conn.Close()
31
+
}
32
+
33
+
select {
34
+
case <-time.After(backoff):
35
+
case <-ctx.Done():
36
+
return ctx.Err()
37
+
}
38
+
39
+
backoff = min(backoff*2, maxBackoff)
40
+
}
41
+
42
+
}
43
+
44
+
func (c *Consumer) stream(ctx context.Context, conn *websocket.Conn, logger *slog.Logger) {
14
45
callbacks := &events.RepoStreamCallbacks{
15
46
RepoCommit: func(event *atproto.SyncSubscribeRepos_Commit) error {
16
47
return c.HandleCommit(ctx, event)
···
19
50
20
51
scheduler := sequential.NewScheduler("asterism", callbacks.EventHandler)
21
52
22
-
return events.HandleRepoStream(ctx, conn, scheduler, logger)
53
+
if err := events.HandleRepoStream(ctx, conn, scheduler, logger); err != nil {
54
+
logger.Warn("stream ended", "err", err)
55
+
}
23
56
}