A fork of the Cocoon PDS but being made more distributed.
event sequencing
Signed-off-by: Will <did:plc:dadhhalkfcq3gucaq25hjqon>
willdot.net
(Jun 6, 2026, 10:32 AM +0100)
bb16c566
1ecd9c61
+55
-33
+1
go.mod
+1
go.mod
+2
go.sum
+2
go.sum
···
9
9
github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA=
10
10
github.com/antlr4-go/antlr/v4 v4.13.0 h1:lxCg3LAv+EUK6t1i0y1V6/SLeUi0eKEKdhQAlS8TVTI=
11
11
github.com/antlr4-go/antlr/v4 v4.13.0/go.mod h1:pfChB/xh/Unjila75QW7+VU4TSnWnnk9UTnmpPaOR2g=
12
+
github.com/avast/retry-go/v4 v4.7.0 h1:yjDs35SlGvKwRNSykujfjdMxMhMQQM0TnIjJaHB+Zio=
13
+
github.com/avast/retry-go/v4 v4.7.0/go.mod h1:ZMPDa3sY2bKgpLtap9JRUgk2yTAba7cgiFhqxY2Sg6Q=
12
14
github.com/aws/aws-sdk-go v1.55.7 h1:UJrkFq7es5CShfBwlWAC8DA077vp8PyVbQd3lqLiztE=
13
15
github.com/aws/aws-sdk-go v1.55.7/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
14
16
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
+52
-33
server/persist.go
+52
-33
server/persist.go
···
3
3
import (
4
4
"bytes"
5
5
"context"
6
+
"errors"
6
7
"fmt"
7
8
"log/slog"
8
9
"sync"
9
10
"time"
10
11
12
+
"github.com/avast/retry-go/v4"
11
13
"github.com/bluesky-social/indigo/api/atproto"
12
14
"github.com/bluesky-social/indigo/events"
13
15
indigomodels "github.com/bluesky-social/indigo/models"
···
60
62
slog.Error("fetching most recent event record", "error", err)
61
63
rec.Seq = time.Now().Unix()
62
64
}
63
-
rec.Seq++
65
+
66
+
// if the error on inserting the event record is a constraint error, it means that
67
+
// another event has been stored since getting the last sequence number. In that case
68
+
// retry which will increase the sequence number again and hopefully work. Any others
69
+
// can error out.
70
+
retryIfFunc := func(err error) bool {
71
+
return errors.Is(gorm.ErrCheckConstraintViolated, err)
72
+
}
73
+
err := retry.Do(func() error {
74
+
rec.Seq++
75
+
76
+
var did string
77
+
var evtType string
78
+
79
+
switch {
80
+
case e.RepoCommit != nil:
81
+
e.RepoCommit.Seq = rec.Seq
82
+
did = e.RepoCommit.Repo
83
+
evtType = "commit"
84
+
case e.RepoSync != nil:
85
+
e.RepoSync.Seq = rec.Seq
86
+
did = e.RepoSync.Did
87
+
evtType = "sync"
88
+
case e.RepoIdentity != nil:
89
+
e.RepoIdentity.Seq = rec.Seq
90
+
did = e.RepoIdentity.Did
91
+
evtType = "identity"
92
+
case e.RepoAccount != nil:
93
+
e.RepoAccount.Seq = rec.Seq
94
+
did = e.RepoAccount.Did
95
+
evtType = "account"
96
+
default:
97
+
return fmt.Errorf("unknown event type")
98
+
}
64
99
65
-
var did string
66
-
var evtType string
100
+
data, err := serializeEvent(e)
101
+
if err != nil {
102
+
return fmt.Errorf("failed to serialize event: %w", err)
103
+
}
67
104
68
-
switch {
69
-
case e.RepoCommit != nil:
70
-
e.RepoCommit.Seq = rec.Seq
71
-
did = e.RepoCommit.Repo
72
-
evtType = "commit"
73
-
case e.RepoSync != nil:
74
-
e.RepoSync.Seq = rec.Seq
75
-
did = e.RepoSync.Did
76
-
evtType = "sync"
77
-
case e.RepoIdentity != nil:
78
-
e.RepoIdentity.Seq = rec.Seq
79
-
did = e.RepoIdentity.Did
80
-
evtType = "identity"
81
-
case e.RepoAccount != nil:
82
-
e.RepoAccount.Seq = rec.Seq
83
-
did = e.RepoAccount.Did
84
-
evtType = "account"
85
-
default:
86
-
return fmt.Errorf("unknown event type")
87
-
}
105
+
rec.CreatedAt = time.Now()
106
+
rec.Did = did
107
+
rec.Type = evtType
108
+
rec.Data = data
88
109
89
-
data, err := serializeEvent(e)
90
-
if err != nil {
91
-
return fmt.Errorf("failed to serialize event: %w", err)
92
-
}
110
+
err = p.Db.Create(rec).Error
111
+
if err != nil {
112
+
return fmt.Errorf("failed to persist event: %w", err)
113
+
}
93
114
94
-
rec.CreatedAt = time.Now()
95
-
rec.Did = did
96
-
rec.Type = evtType
97
-
rec.Data = data
115
+
return nil
116
+
}, retry.RetryIf(retryIfFunc))
98
117
99
-
if err := p.Db.Create(rec).Error; err != nil {
100
-
return fmt.Errorf("failed to persist event: %w", err)
118
+
if err != nil {
119
+
return err
101
120
}
102
121
103
122
if p.Broadcast != nil {