A diet Tap option for handling atproto sync
Tap: fix repo commit metadata tracking (#1280)
Fixes a few issues in Tap. The most important is that we were not
updating prev data & rev when we didn't need to process any ops from a
commit. This would cause commits to appear as disjunctions in commit
history.
This also rearranges some logic in `ProcessCommit` which would cause us
to mark a repo "desynced" if an event came in while resyncing rather
than adding that event to the resync buffer.
For events that we add to the resync buffer, we hold off checking that
they play neatly on the current state of the repo (through prevData)
until we actually go to play them back.
authored by
Daniel Holmgren
and committed by
GitHub
(Jan 26, 2026, 2:18 PM -0600)
e4b2f9cf
151d4c57
+78
-71
+14
-33
cmd/tap/event_manager.go
+14
-33
cmd/tap/event_manager.go
···
3
3
import (
4
4
"context"
5
5
"encoding/json"
6
-
"fmt"
7
6
"log/slog"
8
7
"sync"
9
8
"sync/atomic"
···
151
150
}
152
151
153
152
func (em *EventManager) AddCommit(ctx context.Context, commit *Commit, dbCallback DBCallback) error {
153
+
if len(commit.Ops) == 0 {
154
+
return updateRepoCommitMeta(em.db, commit)
155
+
}
156
+
154
157
evts := make([]*RecordEvt, 0, len(commit.Ops))
155
158
156
159
for _, op := range commit.Ops {
···
172
175
return err
173
176
}
174
177
175
-
return tx.Model(&models.Repo{}).
176
-
Where("did = ?", commit.Did).
177
-
Updates(map[string]interface{}{
178
-
"rev": commit.Rev,
179
-
"prev_data": commit.DataCid,
180
-
}).Error
178
+
return updateRepoCommitMeta(tx, commit)
181
179
})
182
180
}
183
181
182
+
func updateRepoCommitMeta(dbOrTx *gorm.DB, commit *Commit) error {
183
+
return dbOrTx.Model(&models.Repo{}).
184
+
Where("did = ?", commit.Did).
185
+
Updates(map[string]interface{}{
186
+
"rev": commit.Rev,
187
+
"prev_data": commit.DataCid,
188
+
}).Error
189
+
}
190
+
184
191
func (em *EventManager) AddRecordEvents(ctx context.Context, evts []*RecordEvt, live bool, dbCallback DBCallback) error {
185
192
toPut := make([]*models.RepoRecord, 0, len(evts))
186
193
toDel := make([]*models.RepoRecord, 0)
···
322
329
Data: string(jsonData),
323
330
}).Error
324
331
}
325
-
326
-
func (em *EventManager) drainResyncBuffer(ctx context.Context, did string) error {
327
-
var bufferedEvts []models.ResyncBuffer
328
-
if err := em.db.WithContext(ctx).Where("did = ?", did).Order("id ASC").Find(&bufferedEvts).Error; err != nil {
329
-
return fmt.Errorf("failed to load buffered events: %w", err)
330
-
}
331
-
332
-
if len(bufferedEvts) == 0 {
333
-
return nil
334
-
}
335
-
336
-
for _, evt := range bufferedEvts {
337
-
var commit Commit
338
-
if err := json.Unmarshal([]byte(evt.Data), &commit); err != nil {
339
-
return fmt.Errorf("failed to unmarshal buffered event: %w", err)
340
-
}
341
-
342
-
if err := em.AddCommit(ctx, &commit, func(tx *gorm.DB) error {
343
-
return tx.Delete(&models.ResyncBuffer{}, "id = ?", evt.ID).Error
344
-
}); err != nil {
345
-
return err
346
-
}
347
-
}
348
-
349
-
return nil
350
-
}
+21
-33
cmd/tap/firehose.go
+21
-33
cmd/tap/firehose.go
···
88
88
return nil
89
89
}
90
90
91
-
if evt.PrevData == nil {
92
-
fp.logger.Debug("legacy commit event, skipping prev data check", "did", evt.Repo, "rev", evt.Rev)
93
-
} else if evt.PrevData.String() != curr.PrevData {
94
-
fp.logger.Warn("repo state desynchronized", "did", evt.Repo, "rev", evt.Rev)
95
-
// gets picked up by resync workers
96
-
if err := fp.repos.UpdateRepoState(ctx, evt.Repo, models.RepoStateDesynchronized); err != nil {
97
-
fp.logger.Error("failed to update repo state to desynchronized", "did", evt.Repo, "error", err)
98
-
return err
99
-
}
100
-
return nil
101
-
}
102
-
103
-
commit, err := fp.validateCommit(ctx, evt)
91
+
commit, err := fp.validateCommitAndFilterOps(ctx, evt)
104
92
if err != nil {
105
93
fp.logger.Error("failed to parse operations", "did", evt.Repo, "error", err)
106
94
return err
107
95
}
108
96
109
-
// filter ops to only matching collections after validation (since all ops are necessary for commit validation)
110
-
filteredOps := []CommitOp{}
111
-
for _, op := range commit.Ops {
112
-
if matchesCollection(op.Collection, fp.collectionFilters) {
113
-
filteredOps = append(filteredOps, op)
114
-
}
115
-
}
116
-
if len(filteredOps) == 0 {
97
+
if curr.State == models.RepoStateResyncing {
117
98
firehoseEventsSkipped.Inc()
118
-
return nil
99
+
return fp.events.addToResyncBuffer(ctx, commit)
119
100
}
120
-
commit.Ops = filteredOps
121
101
122
-
if curr.State == models.RepoStateResyncing {
123
-
if err := fp.events.addToResyncBuffer(ctx, commit); err != nil {
124
-
fp.logger.Error("failed to buffer commit", "did", evt.Repo, "error", err)
102
+
if evt.PrevData == nil {
103
+
fp.logger.Debug("legacy commit event, skipping prev data check", "did", evt.Repo, "rev", evt.Rev)
104
+
} else if evt.PrevData.String() != curr.PrevData {
105
+
fp.logger.Warn("repo state desynchronized", "did", evt.Repo, "rev", evt.Rev)
106
+
// gets picked up by resync workers
107
+
if err := fp.repos.UpdateRepoState(ctx, evt.Repo, models.RepoStateDesynchronized); err != nil {
108
+
fp.logger.Error("failed to update repo state to desynchronized", "did", evt.Repo, "error", err)
125
109
return err
126
110
}
111
+
return nil
127
112
}
128
113
129
114
if err := fp.events.AddCommit(ctx, commit, func(tx *gorm.DB) error {
···
137
122
return nil
138
123
}
139
124
140
-
func (fp *FirehoseProcessor) validateCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) (*Commit, error) {
125
+
func (fp *FirehoseProcessor) validateCommitAndFilterOps(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) (*Commit, error) {
141
126
if err := repo.VerifyCommitSignature(ctx, fp.repos.IdDir, evt); err != nil {
142
127
return nil, err
143
128
}
···
147
132
return nil, err
148
133
}
149
134
150
-
var parsedOps []CommitOp
135
+
parsedOps := make([]CommitOp, 0)
151
136
152
137
for _, op := range evt.Ops {
153
138
collection, rkey, err := syntax.ParseRepoPath(op.Path)
···
182
167
parsed.Record = record
183
168
}
184
169
185
-
parsedOps = append(parsedOps, parsed)
170
+
if matchesCollection(parsed.Collection, fp.collectionFilters) {
171
+
parsedOps = append(parsedOps, parsed)
172
+
}
186
173
}
187
174
188
175
repoCommit, err := r.Commit()
···
191
178
}
192
179
193
180
commit := &Commit{
194
-
Did: evt.Repo,
195
-
Rev: repoCommit.Rev,
196
-
DataCid: repoCommit.Data.String(),
197
-
Ops: parsedOps,
181
+
Did: evt.Repo,
182
+
Rev: repoCommit.Rev,
183
+
DataCid: repoCommit.Data.String(),
184
+
PrevData: evt.PrevData.String(),
185
+
Ops: parsedOps,
198
186
}
199
187
200
188
return commit, nil
+38
-1
cmd/tap/resyncer.go
+38
-1
cmd/tap/resyncer.go
···
3
3
import (
4
4
"bytes"
5
5
"context"
6
+
"encoding/json"
6
7
"fmt"
7
8
"log/slog"
8
9
"net/http"
···
119
120
resyncsCompleted.Inc()
120
121
resyncDuration.Observe(time.Since(startTime).Seconds())
121
122
122
-
if err := r.events.drainResyncBuffer(ctx, did); err != nil {
123
+
if err := r.drainResyncBuffer(ctx, did); err != nil {
123
124
r.logger.Error("failed to drain resync buffer events", "did", did, "error", err)
124
125
}
125
126
···
346
347
Where("state = ?", models.RepoStateResyncing).
347
348
Update("state", models.RepoStateDesynchronized).Error
348
349
}
350
+
351
+
func (r *Resyncer) drainResyncBuffer(ctx context.Context, did string) error {
352
+
var bufferedEvts []models.ResyncBuffer
353
+
if err := r.events.db.WithContext(ctx).Where("did = ?", did).Order("id ASC").Find(&bufferedEvts).Error; err != nil {
354
+
return fmt.Errorf("failed to load buffered events: %w", err)
355
+
}
356
+
357
+
if len(bufferedEvts) == 0 {
358
+
return nil
359
+
}
360
+
361
+
curr, err := r.repos.GetRepoState(ctx, did)
362
+
if err != nil {
363
+
return fmt.Errorf("failed to get repo state: %w", err)
364
+
}
365
+
366
+
for _, evt := range bufferedEvts {
367
+
var commit Commit
368
+
if err := json.Unmarshal([]byte(evt.Data), &commit); err != nil {
369
+
return fmt.Errorf("failed to unmarshal buffered event: %w", err)
370
+
}
371
+
372
+
// if this commit doesn't stack neatly on current state of tracked repo then we skip
373
+
if commit.PrevData != curr.PrevData {
374
+
continue
375
+
}
376
+
377
+
if err := r.events.AddCommit(ctx, &commit, func(tx *gorm.DB) error {
378
+
return tx.Delete(&models.ResyncBuffer{}, "id = ?", evt.ID).Error
379
+
}); err != nil {
380
+
return err
381
+
}
382
+
}
383
+
384
+
return nil
385
+
}
+5
-4
cmd/tap/types.go
+5
-4
cmd/tap/types.go
···
15
15
)
16
16
17
17
type Commit struct {
18
-
Did string `json:"did"`
19
-
Rev string `json:"rev"`
20
-
DataCid string `json:"data_cid"`
21
-
Ops []CommitOp `json:"ops"`
18
+
Did string `json:"did"`
19
+
Rev string `json:"rev"`
20
+
DataCid string `json:"data_cid"`
21
+
PrevData string `json:"prev_data"`
22
+
Ops []CommitOp `json:"ops"`
22
23
}
23
24
24
25
type CommitOp struct {