Monorepo for Tangled
appview,knotserver,spindle/ingester: no more 2-day past cap
Lewis: May this revision serve well! <lewis@tangled.org>
+109
-38
-5
appview/ingester.go
-5
appview/ingester.go
+28
-23
jetstream/jetstream.go
+28
-23
jetstream/jetstream.go
···
7
7
"os"
8
8
"os/signal"
9
9
"sync"
10
+
"sync/atomic"
10
11
"syscall"
11
12
"time"
12
13
···
34
35
db DB
35
36
waitForDid bool
36
37
mu sync.RWMutex
38
+
39
+
lastSeenUs atomic.Int64
37
40
38
41
cancel context.CancelFunc
39
42
cancelMu sync.Mutex
···
71
74
// since this closure references j.WantedDids; it should auto-update
72
75
// existing instances of the closure when j.WantedDids is mutated
73
76
return func(ctx context.Context, evt *models.Event) error {
74
-
75
77
j.mu.RLock()
76
78
// empty filter => all dids allowed
77
79
matches := len(j.wantedDids) == 0
···
82
84
}
83
85
j.mu.RUnlock()
84
86
87
+
var err error
85
88
if matches {
86
-
return processFunc(ctx, evt)
87
-
} else {
88
-
return nil
89
+
err = processFunc(ctx, evt)
89
90
}
91
+
92
+
j.lastSeenUs.Store(evt.TimeUS + 1)
93
+
return err
90
94
}
91
95
}
92
96
···
113
117
}
114
118
115
119
// StartJetstream starts the jetstream client and processes events using the provided processFunc.
116
-
// The caller is responsible for saving the last time_us to the database (just use your db.UpdateLastTimeUs).
120
+
// The client persists the last time_us cursor itself via the DB it was constructed with.
117
121
func (j *JetstreamClient) StartJetstream(ctx context.Context, processFunc func(context.Context, *models.Event) error) error {
118
122
logger := j.l
119
123
···
151
155
func (j *JetstreamClient) connectAndRead(ctx context.Context) {
152
156
l := log.FromContext(ctx)
153
157
for {
154
-
cursor := j.getLastTimeUs(ctx)
158
+
cursor := j.resumeCursor(ctx)
155
159
156
160
connCtx, cancel := context.WithCancel(ctx)
157
161
j.cancelMu.Lock()
···
185
189
case <-ctx.Done():
186
190
return
187
191
case <-ticker.C:
188
-
j.db.SaveLastTimeUs(time.Now().UnixMicro())
192
+
if seen := j.lastSeenUs.Load(); seen != 0 {
193
+
if err := j.db.SaveLastTimeUs(seen); err != nil {
194
+
log.FromContext(ctx).Error("failed to save cursor", "error", err)
195
+
}
196
+
}
189
197
}
190
198
}
191
199
}
192
200
201
+
func (j *JetstreamClient) resumeCursor(ctx context.Context) *int64 {
202
+
if seen := j.lastSeenUs.Load(); seen != 0 {
203
+
return &seen
204
+
}
205
+
return j.getLastTimeUs(ctx)
206
+
}
207
+
193
208
func (j *JetstreamClient) getLastTimeUs(ctx context.Context) *int64 {
194
209
l := log.FromContext(ctx)
195
210
lastTimeUs, err := j.db.GetLastTimeUs()
196
211
if err != nil {
197
212
l.Warn("couldn't get last time us, starting from now", "error", err)
198
213
lastTimeUs = time.Now().UnixMicro()
199
-
err = j.db.SaveLastTimeUs(lastTimeUs)
200
-
if err != nil {
201
-
l.Error("failed to save last time us", "error", err)
202
-
}
203
-
}
204
-
205
-
// If last time is older than 2 days, start from now
206
-
if time.Now().UnixMicro()-lastTimeUs > 2*24*60*60*1000*1000 {
207
-
lastTimeUs = time.Now().UnixMicro()
208
-
l.Warn("last time us is older than 2 days; discarding that and starting from now")
209
-
err = j.db.SaveLastTimeUs(lastTimeUs)
210
-
if err != nil {
214
+
if err = j.db.SaveLastTimeUs(lastTimeUs); err != nil {
211
215
l.Error("failed to save last time us", "error", err)
212
216
}
213
217
}
···
234
238
sig := <-sigChan
235
239
j.l.Info("Received signal, initiating graceful shutdown", "signal", sig)
236
240
237
-
lastTimeUs := time.Now().UnixMicro()
238
-
if err := j.db.SaveLastTimeUs(lastTimeUs); err != nil {
239
-
j.l.Error("Failed to save last time during shutdown", "error", err)
241
+
if seen := j.lastSeenUs.Load(); seen != 0 {
242
+
if err := j.db.SaveLastTimeUs(seen); err != nil {
243
+
j.l.Error("Failed to save last time during shutdown", "error", err)
244
+
}
245
+
j.l.Info("Saved lastTimeUs before shutdown", "lastTimeUs", seen)
240
246
}
241
-
j.l.Info("Saved lastTimeUs before shutdown", "lastTimeUs", lastTimeUs)
242
247
243
248
j.cancelMu.Lock()
244
249
if j.cancel != nil {
+81
jetstream/jetstream_test.go
+81
jetstream/jetstream_test.go
···
1
+
package jetstream
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"testing"
7
+
"time"
8
+
9
+
"github.com/bluesky-social/jetstream/pkg/models"
10
+
)
11
+
12
+
type fakeCursorDB struct {
13
+
saved int64
14
+
savedErr error
15
+
}
16
+
17
+
func (f *fakeCursorDB) GetLastTimeUs() (int64, error) { return f.saved, f.savedErr }
18
+
func (f *fakeCursorDB) SaveLastTimeUs(int64) error { return nil }
19
+
20
+
func cursorFor(db DB) int64 {
21
+
j := &JetstreamClient{db: db, wantedDids: make(Set[string])}
22
+
return *j.getLastTimeUs(context.Background())
23
+
}
24
+
25
+
const twoDaysUs = int64(2 * 24 * 60 * 60 * 1000 * 1000)
26
+
27
+
func TestStaleCursorIsHonored(t *testing.T) {
28
+
old := time.Now().UnixMicro() - 5*twoDaysUs
29
+
if got := cursorFor(&fakeCursorDB{saved: old}); got != old {
30
+
t.Fatalf("stale cursor must be honored, not snapped forward: got %d, want %d", got, old)
31
+
}
32
+
}
33
+
34
+
func TestZeroCursorIsHonored(t *testing.T) {
35
+
if got := cursorFor(&fakeCursorDB{saved: 0}); got != 0 {
36
+
t.Fatalf("cursor 0 must replay from the start: got %d, want 0", got)
37
+
}
38
+
}
39
+
40
+
func TestMissingCursorStartsFromNow(t *testing.T) {
41
+
before := time.Now().UnixMicro()
42
+
got := cursorFor(&fakeCursorDB{savedErr: errors.New("no row")})
43
+
after := time.Now().UnixMicro()
44
+
if got < before || got > after {
45
+
t.Fatalf("missing cursor must start from now: got %d, want within [%d,%d]", got, before, after)
46
+
}
47
+
}
48
+
49
+
func TestLastSeenTracksEveryEventPreFilter(t *testing.T) {
50
+
j := &JetstreamClient{wantedDids: Set[string]{"did:plc:boltless": {}}}
51
+
wrapped := j.withDidFilter(func(context.Context, *models.Event) error { return nil })
52
+
53
+
_ = wrapped(context.Background(), &models.Event{Did: "did:plc:akshay", TimeUS: 100})
54
+
if got := j.lastSeenUs.Load(); got != 101 {
55
+
t.Fatalf("filtered-out event must still advance last-seen: got %d, want 101", got)
56
+
}
57
+
58
+
_ = wrapped(context.Background(), &models.Event{Did: "did:plc:boltless", TimeUS: 200})
59
+
if got := j.lastSeenUs.Load(); got != 201 {
60
+
t.Fatalf("matching event must advance last-seen: got %d, want 201", got)
61
+
}
62
+
}
63
+
64
+
func TestLastSeenAdvancesAfterProcessing(t *testing.T) {
65
+
j := &JetstreamClient{wantedDids: make(Set[string])}
66
+
67
+
var seenDuringProcess int64
68
+
wrapped := j.withDidFilter(func(context.Context, *models.Event) error {
69
+
seenDuringProcess = j.lastSeenUs.Load()
70
+
return nil
71
+
})
72
+
73
+
_ = wrapped(context.Background(), &models.Event{Did: "did:plc:boltless", TimeUS: 500})
74
+
75
+
if seenDuringProcess == 501 {
76
+
t.Fatal("cursor advanced before the event finished processing; a crash mid-process would skip it")
77
+
}
78
+
if got := j.lastSeenUs.Load(); got != 501 {
79
+
t.Fatalf("cursor must advance once processing returns: got %d, want 501", got)
80
+
}
81
+
}
-5
knotserver/ingester.go
-5
knotserver/ingester.go
-5
spindle/ingester.go
-5
spindle/ingester.go
···
37
37
s.l.Warn("failed to process message, skipping", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err)
38
38
}
39
39
40
-
lastTimeUs := e.TimeUS + 1
41
-
if saveErr := s.db.SaveLastTimeUs(lastTimeUs); saveErr != nil {
42
-
s.l.Error("failed to save cursor", "err", saveErr)
43
-
}
44
-
45
40
return nil
46
41
}
47
42
}