alice.pds.demo.boltless.dev
This repository has no description
knotserver/jetstream: preserve last time_us even on err'd exit
Also reduces last time_us discard window down to 2 days.
author
Anirudh Oppiliappan
date
(Feb 24, 2025, 9:56 PM +0200)
commit
80520d2c
80520d2c6a5a3fad33777803a55f3c20851b5dda
parent
509296a7
509296a78834ed8fd050ace00661b36854957718
+16
-11
+2
-2
jetstream/jetstream.go
+2
-2
jetstream/jetstream.go
···
141
141
}
142
142
143
143
// If last time is older than a week, start from now
144
-
if time.Now().UnixMicro()-lastTimeUs > 7*24*60*60*1000*1000 {
144
+
if time.Now().UnixMicro()-lastTimeUs > 2*24*60*60*1000*1000 {
145
145
lastTimeUs = time.Now().UnixMicro()
146
-
l.Warn("last time us is older than a week. discarding that and starting from now")
146
+
l.Warn("last time us is older than 2 days; discarding that and starting from now")
147
147
err = j.db.UpdateLastTimeUs(lastTimeUs)
148
148
if err != nil {
149
149
l.Error("failed to save last time us", "error", err)
+14
-9
knotserver/jetstream.go
+14
-9
knotserver/jetstream.go
···
29
29
return nil
30
30
}
31
31
32
-
func (h *Handle) processKnotMember(ctx context.Context, did string, record tangled.KnotMember, eventTime int64) error {
32
+
func (h *Handle) processKnotMember(ctx context.Context, did string, record tangled.KnotMember) error {
33
33
l := log.FromContext(ctx)
34
34
35
35
if record.Domain != h.c.Server.Hostname {
···
58
58
return fmt.Errorf("failed to fetch and add keys: %w", err)
59
59
}
60
60
61
-
lastTimeUs := eventTime + 1
62
-
fmt.Println("lastTimeUs", lastTimeUs)
63
-
if err := h.db.UpdateLastTimeUs(lastTimeUs); err != nil {
64
-
return fmt.Errorf("failed to save last time us: %w", err)
65
-
}
66
-
h.jc.UpdateDids([]string{did})
67
61
return nil
68
62
}
69
63
···
116
110
return nil
117
111
}
118
112
113
+
var err error
114
+
defer func() {
115
+
eventTime := event.TimeUS
116
+
lastTimeUs := eventTime + 1
117
+
fmt.Println("lastTimeUs", lastTimeUs)
118
+
if err := h.db.UpdateLastTimeUs(lastTimeUs); err != nil {
119
+
err = fmt.Errorf("(deferred) failed to save last time us: %w", err)
120
+
}
121
+
h.jc.UpdateDids([]string{did})
122
+
}()
123
+
119
124
raw := json.RawMessage(event.Commit.Record)
120
125
121
126
switch event.Commit.Collection {
···
133
138
if err := json.Unmarshal(raw, &record); err != nil {
134
139
return fmt.Errorf("failed to unmarshal record: %w", err)
135
140
}
136
-
if err := h.processKnotMember(ctx, did, record, event.TimeUS); err != nil {
141
+
if err := h.processKnotMember(ctx, did, record); err != nil {
137
142
return fmt.Errorf("failed to process knot member: %w", err)
138
143
}
139
144
}
140
145
141
-
return nil
146
+
return err
142
147
}