this repo has no description
hepa: use atomics for lastSeq. (#560)
Fixes data race. Was probably harmless in practice, as these values are
best-effort anyway, but keeping the race detector clean and happy when
engaged is highly valuable for maintainability in the long run.
Some comments attached about the field's concurrency.
+20
-10
+4
-3
cmd/hepa/consumer.go
+4
-3
cmd/hepa/consumer.go
···
7
7
"net/http"
8
8
"net/url"
9
9
"strings"
10
+
"sync/atomic"
10
11
11
12
comatproto "github.com/bluesky-social/indigo/api/atproto"
12
13
"github.com/bluesky-social/indigo/atproto/syntax"
···
48
49
49
50
rsc := &events.RepoStreamCallbacks{
50
51
RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {
51
-
s.lastSeq = evt.Seq
52
+
atomic.StoreInt64(&s.lastSeq, evt.Seq)
52
53
return s.HandleRepoCommit(ctx, evt)
53
54
},
54
55
RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error {
55
-
s.lastSeq = evt.Seq
56
+
atomic.StoreInt64(&s.lastSeq, evt.Seq)
56
57
did, err := syntax.ParseDID(evt.Did)
57
58
if err != nil {
58
59
s.logger.Error("bad DID in RepoHandle event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
···
64
65
return nil
65
66
},
66
67
RepoMigrate: func(evt *comatproto.SyncSubscribeRepos_Migrate) error {
67
-
s.lastSeq = evt.Seq
68
+
atomic.StoreInt64(&s.lastSeq, evt.Seq)
68
69
did, err := syntax.ParseDID(evt.Did)
69
70
if err != nil {
70
71
s.logger.Error("bad DID in RepoMigrate event", "did", evt.Did, "seq", evt.Seq, "err", err)
+16
-7
cmd/hepa/server.go
+16
-7
cmd/hepa/server.go
···
7
7
"net/http"
8
8
"os"
9
9
"strings"
10
+
"sync/atomic"
10
11
"time"
11
12
12
13
comatproto "github.com/bluesky-social/indigo/api/atproto"
···
30
31
logger *slog.Logger
31
32
engine *automod.Engine
32
33
rdb *redis.Client
34
+
35
+
// lastSeq is the most recent event sequence number we've received and begun to handle.
36
+
// This number is periodically persisted to redis, if redis is present.
37
+
// The value is best-effort (the stream handling itself is concurrent, so event numbers may not be monotonic),
38
+
// but nonetheless, you must use atomics when updating or reading this (to avoid data races).
33
39
lastSeq int64
34
40
}
35
41
···
233
239
if s.rdb == nil {
234
240
return nil
235
241
}
236
-
if s.lastSeq <= 0 {
242
+
lastSeq := atomic.LoadInt64(&s.lastSeq)
243
+
if lastSeq <= 0 {
237
244
return nil
238
245
}
239
-
err := s.rdb.Set(ctx, cursorKey, s.lastSeq, 14*24*time.Hour).Err()
246
+
err := s.rdb.Set(ctx, cursorKey, lastSeq, 14*24*time.Hour).Err()
240
247
return err
241
248
}
242
249
···
289
296
for {
290
297
select {
291
298
case <-ctx.Done():
292
-
if s.lastSeq >= 1 {
293
-
s.logger.Info("persisting final cursor seq value", "seq", s.lastSeq)
299
+
lastSeq := atomic.LoadInt64(&s.lastSeq)
300
+
if lastSeq >= 1 {
301
+
s.logger.Info("persisting final cursor seq value", "seq", lastSeq)
294
302
err := s.PersistCursor(ctx)
295
303
if err != nil {
296
-
s.logger.Error("failed to persist cursor", "err", err, "seq", s.lastSeq)
304
+
s.logger.Error("failed to persist cursor", "err", err, "seq", lastSeq)
297
305
}
298
306
}
299
307
return nil
300
308
case <-ticker.C:
301
-
if s.lastSeq >= 1 {
309
+
lastSeq := atomic.LoadInt64(&s.lastSeq)
310
+
if lastSeq >= 1 {
302
311
err := s.PersistCursor(ctx)
303
312
if err != nil {
304
-
s.logger.Error("failed to persist cursor", "err", err, "seq", s.lastSeq)
313
+
s.logger.Error("failed to persist cursor", "err", err, "seq", lastSeq)
305
314
}
306
315
}
307
316
}