negentropy.mx raw
1 // Package negentropy implements the negentropy set reconciliation protocol
2 // for efficient event sync between relays.
3 //
4 // The protocol works by exchanging XOR fingerprints over sorted ranges of
5 // (timestamp, id) pairs. Matching fingerprints mean the range is identical;
6 // mismatched ranges are subdivided recursively until individual items are
7 // exchanged.
8 package negentropy
9
10 import (
11 "git.smesh.lol/moxie/pkg/mxutil"
12 "bytes"
13 "crypto/sha256"
14 "math"
15
16 "git.smesh.lol/nostr/pkg/event"
17 "git.smesh.lol/nostr/pkg/filter"
18 "git.smesh.lol/morly/pkg/store"
19 )
20
21 // Item is a (timestamp, id) pair for set reconciliation.
22 type Item struct {
23 Timestamp int64
24 ID []byte // 32 bytes
25 }
26
27 func compareItems(a, b Item) (n int32) {
28 if a.Timestamp != b.Timestamp {
29 if a.Timestamp < b.Timestamp {
30 return -1
31 }
32 return 1
33 }
34 return bytes.Compare(a.ID, b.ID)
35 }
36
37 // ItemsFromEvents extracts sorted items from events.
38 func ItemsFromEvents(events []*event.E) (is []Item) {
39 items := []Item{:len(events)}
40 for i, ev := range events {
41 id := []byte{:32}
42 copy(id, ev.ID)
43 items[i] = Item{Timestamp: ev.CreatedAt, ID: id}
44 }
45 sortItems(items)
46 return items
47 }
48
49 func sortItems(items []Item) {
50 // Moxie's sort package has no Slice; insertion sort is the tree's own
51 // shape (see pkg/nostr/filter and cmd/mxcover). Sorting ascends by
52 // (Timestamp, ID), matching compareItems.
53 for i := int32(1); i < len(items); i++ {
54 for j := i; j > 0 && compareItems(items[j], items[j-1]) < 0; j-- {
55 items[j], items[j-1] = items[j-1], items[j]
56 }
57 }
58 }
59
60 // Fingerprint computes an XOR fingerprint over a range of items.
61 func Fingerprint(items []Item) (v [32]byte) {
62 var fp [32]byte
63 for _, it := range items {
64 for j := 0; j < 32 && j < len(it.ID); j++ {
65 fp[j] ^= it.ID[j]
66 }
67 }
68 return fp
69 }
70
71 // Diff computes set differences between two sorted item lists.
72 // Returns have (items in local but not remote) and need (items in remote but not local).
73 func Diff(local, remote []Item) (have, need []Item) {
74 i, j := 0, 0
75 for i < len(local) && j < len(remote) {
76 cmp := compareItems(local[i], remote[j])
77 if cmp < 0 {
78 have = mxutil.Ensure(have, 1)
79 have = push(have, local[i])
80 i++
81 } else if cmp > 0 {
82 need = mxutil.Ensure(need, 1)
83 need = push(need, remote[j])
84 j++
85 } else {
86 i++
87 j++
88 }
89 }
90 for ; i < len(local); i++ {
91 have = mxutil.Ensure(have, 1)
92 have = push(have, local[i])
93 }
94 for ; j < len(remote); j++ {
95 need = mxutil.Ensure(need, 1)
96 need = push(need, remote[j])
97 }
98 return
99 }
100
101 // Reconciler performs fingerprint-based reconciliation.
102 type Reconciler struct {
103 items []Item
104 }
105
106 // NewReconciler creates a reconciler from a sorted item list.
107 func NewReconciler(items []Item) (r *Reconciler) {
108 sortItems(items)
109 return &Reconciler{items: items}
110 }
111
112 // Range represents a segment with a fingerprint.
113 type Range struct {
114 UpperTimestamp int64
115 UpperID []byte
116 Fingerprint [32]byte
117 Count int32
118 }
119
120 // Split divides the item set into n ranges with fingerprints.
121 func (r *Reconciler) Split(n int32) (rs []Range) {
122 if n <= 0 || len(r.items) == 0 {
123 return nil
124 }
125 if n > len(r.items) {
126 n = len(r.items)
127 }
128 segSize := len(r.items) / n
129 ranges := []Range{:0:n}
130 for i := 0; i < n; i++ {
131 lo := i * segSize
132 hi := (i + 1) * segSize
133 if i == n-1 {
134 hi = len(r.items)
135 }
136 upper := r.items[hi-1]
137 ranges = push(ranges, Range{
138 UpperTimestamp: upper.Timestamp,
139 UpperID: upper.ID,
140 Fingerprint: Fingerprint(r.items[lo:hi]),
141 Count: hi - lo,
142 })
143 }
144 return ranges
145 }
146
147 // FindMismatches compares local ranges against remote ranges and
148 // returns indices of ranges that differ.
149 func FindMismatches(local, remote []Range) (ss []int32) {
150 n := len(local)
151 if len(remote) < n {
152 n = len(remote)
153 }
154 var mismatches []int32
155 for i := 0; i < n; i++ {
156 if local[i].Fingerprint != remote[i].Fingerprint {
157 mismatches = mxutil.Ensure(mismatches, 1)
158 mismatches = push(mismatches, i)
159 }
160 }
161 return mismatches
162 }
163
164 // Syncer orchestrates sync between local store and a remote item set.
165 type Syncer struct {
166 store *store.Engine
167 }
168
169 // NewSyncer creates a syncer.
170 func NewSyncer(s *store.Engine) (sv *Syncer) { return &Syncer{store: s} }
171
172 // LocalItems returns all (timestamp, id) pairs from the local store.
173 func (s *Syncer) LocalItems() (is []Item) {
174 f := &filter.F{}
175 events, err := s.store.QueryEvents(f)
176 if err != nil {
177 return nil
178 }
179 return ItemsFromEvents(events)
180 }
181
182 // FindNeeded compares local items against remote items and returns
183 // IDs of events we need to fetch.
184 func (s *Syncer) FindNeeded(remoteItems []Item) (ss [][]byte) {
185 local := s.LocalItems()
186 _, need := Diff(local, remoteItems)
187 ids := [][]byte{:len(need)}
188 for i, it := range need {
189 ids[i] = it.ID
190 }
191 return ids
192 }
193
194 // FindHave compares local items against remote items and returns
195 // events we have that the remote doesn't.
196 func (s *Syncer) FindHave(remoteItems []Item) (ss []*event.E) {
197 local := s.LocalItems()
198 have, _ := Diff(local, remoteItems)
199 var events []*event.E
200 for _, it := range have {
201 ev, err := s.store.GetByID(it.ID)
202 if err == nil && ev != nil {
203 events = mxutil.Ensure(events, 1)
204 events = push(events, ev)
205 }
206 }
207 return events
208 }
209
210 func sha256Hash(data []byte) (buf []byte) {
211 h := sha256.Sum256(data)
212 buf = []byte{:32}
213 copy(buf, h[:])
214 return
215 }
216
217 // EstimateRanges returns the optimal number of ranges for a given item count.
218 func EstimateRanges(n int32) (nv int32) {
219 if n <= 0 {
220 return 0
221 }
222 r := int32(math.Ceil(math.Sqrt(float64(n))))
223 if r < 2 {
224 return 2
225 }
226 if r > 128 {
227 return 128
228 }
229 return r
230 }
231