{{ message }}
Repository navigation
Expand file tree
/
Copy pathsync.js
More file actions
317 lines (287 loc) · 9.7 KB
/
Copy pathsync.js
File metadata and controls
317 lines (287 loc) · 9.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
import PQueue from 'p-queue'
import { EventEmitter } from 'events'
import { TimeoutController } from 'timeout-abort-controller'
import pathJoin from './utils/path-join.js'
import { Entry } from './oplog/index.js'
const DefaultTimeout = 30000 // 30 seconds
/**
* @module Sync
* @description
* The Sync Protocol for OrbitDB synchronizes the database operations {@link module:Log} between multiple peers.
*
* The Sync Protocol sends and receives heads between multiple peers,
* both when opening a database and when a database is updated, ie.
* new entries are appended to the log.
*
* When Sync is started, a peer subscribes to a pubsub topic of the log's id.
* Upon subscribing to the topic, peers already connected to the topic receive
* the subscription message and "dial" the subscribing peer using a libp2p
* custom protocol. Once connected to the subscribing peer on a direct
* peer-to-peer connection, the dialing peer and the subscribing peer exchange
* the heads of the Log each peer currently has. Once completed, the peers have
* the same "local state".
*
* Once the initial sync has completed, peers notify one another of updates to
* the log, ie. updates to the database, using the initially opened pubsub
* topic subscription. A peer with new heads broadcasts changes to other peers
* by publishing the updated heads to the pubsub topic. Peers subscribed to the
* same topic will then receive the update and will update their log's state,
* the heads, accordingly.
*
* The Sync Protocol is eventually consistent. It guarantees that once all
* messages have been sent and received, peers will observe the same log state
* and values. The Sync Protocol does not guarantee the order in which messages
* are received or even that a message is recieved at all, nor any timing on
* when messages are received.
*
* @example
* // Using defaults
* const sync = await Sync({ ipfs, log, onSynced: (peerId, heads) => ... })
*
* @example
* // Using all parameters
* const sync = await Sync({ ipfs, log, events, onSynced: (peerId, heads) => ..., start: false })
* sync.events.on('join', (peerId, heads) => ...)
* sync.events.on('leave', (peerId) => ...)
* sync.events.on('error', (err) => ...)
* await sync.start()
*/
/**
* Creates a Sync instance for sychronizing logs between multiple peers.
*
* @function
* @param {Object} params One or more parameters for configuring Sync.
* @param {IPFS} params.ipfs An IPFS instance.
* @param {Log} params.log The log instance to sync.
* @param {EventEmitter} [params.events] An event emitter to use. Events
* emitted are 'join', 'leave' and 'error'. If the parameter is not provided,
* an EventEmitter will be created.
* @param {onSynced} [params.onSynced] A callback function that is called after
* the peer has received heads from another peer.
* @param {Boolean} [params.start] True if sync should start automatically,
* false otherwise. Defaults to true.
* @return {module:Sync~Sync} sync An instance of the Sync Protocol.
* @memberof module:Sync
* @instance
*/
const Sync = async ({ ipfs, log, events, onSynced, start, timeout }) => {
/**
* @namespace module:Sync~Sync
* @description The instance returned by {@link module:Sync}.
*/
/**
* Callback function when new heads have been received from other peers.
* @callback module:Sync~Sync#onSynced
* @param {PeerID} peerId PeerID of the peer who we received heads from
* @param {Entry[]} heads An array of Log entries
*/
/**
* Event fired when when a peer has connected and the exchange of
* heads has been completed.
* @event module:Sync~Sync#join
* @param {PeerID} peerId PeerID of the peer who we received heads from
* @param {Entry[]} heads An array of Log entries
* @example
* sync.events.on('join', (peerID, heads) => ...)
*/
/**
* Event fired when a peer leaves the sync protocol.
* @event module:Sync~Sync#leave
* @param {PeerID} peerId PeerID of the peer who left
* @example
* sync.events.on('leave', (peerID) => ...)
*/
/**
* Event fired when an error occurs.
* @event module:Sync~Sync#error
* @param {Error} error The error that occured
* @example
* sync.events.on('error', (error) => ...)
*/
if (!ipfs) throw new Error('An instance of ipfs is required.')
if (!log) throw new Error('An instance of log is required.')
const libp2p = ipfs.libp2p
const pubsub = ipfs.libp2p.services.pubsub
const address = log.id
const headsSyncAddress = pathJoin('/orbitdb/heads/', address)
const queue = new PQueue({ concurrency: 1 })
/**
* Set of currently connected peers for the log for this Sync instance.
* @name peers
* @†ype Set
* @memberof module:Sync~Sync
* @instance
*/
const peers = new Set()
/**
* Event emitter that emits Sync changes. See Events section for details.
* @†ype EventEmitter
* @memberof module:Sync~Sync
* @instance
*/
events = events || new EventEmitter()
timeout ??= DefaultTimeout
let started = false
const onPeerJoined = async (peerId) => {
const heads = await log.heads()
events.emit('join', peerId, heads)
}
const sendHeads = async (stream) => {
const heads = await log.heads()
for (const { hash } of heads) {
const bytes = await log.storage.get(hash)
if (bytes) {
stream.send(bytes)
}
}
}
const receiveHeads = async (peerId, stream) => {
for await (const value of stream) {
const headBytes = value.subarray()
if (headBytes && onSynced) {
const entry = await Entry.decode(headBytes, log.encryption.replication?.decrypt, log.encryption.data?.decrypt)
await onSynced(entry)
}
}
if (started) {
await onPeerJoined(peerId)
}
}
const handleReceiveHeads = async (stream, connection) => {
const peerId = String(connection.remotePeer)
try {
peers.add(peerId)
await Promise.all([
receiveHeads(peerId, stream),
sendHeads(stream).then(() => stream.close())
])
} catch (e) {
peers.delete(peerId)
events.emit('error', e)
}
}
const handlePeerSubscribed = async (event) => {
const task = async () => {
const { peerId: remotePeer, subscriptions } = event.detail
const peerId = String(remotePeer)
const subscription = subscriptions.find(e => e.topic === address)
if (!subscription) {
return
}
if (subscription.subscribe) {
if (peers.has(peerId)) {
return
}
const timeoutController = new TimeoutController(timeout)
const { signal } = timeoutController
try {
peers.add(peerId)
const stream = await libp2p.dialProtocol(remotePeer, headsSyncAddress, { signal })
await Promise.all([
receiveHeads(peerId, stream),
sendHeads(stream).then(() => stream.close())
])
} catch (e) {
peers.delete(peerId)
if (e.name === 'UnsupportedProtocolError') {
// Skip peer, they don't have this database currently
} else {
if (e.name === 'AbortError' && e.type == null) {
e.type = 'aborted'
}
events.emit('error', e)
}
} finally {
if (timeoutController) {
timeoutController.clear()
}
}
} else {
peers.delete(peerId)
events.emit('leave', peerId)
}
}
queue.add(task)
}
const handleUpdateMessage = async message => {
const { topic, data } = message.detail
const task = async () => {
try {
if (data && onSynced) {
const entry = await Entry.decode(data, log.encryption.replication?.decrypt, log.encryption.data?.decrypt)
await onSynced(entry)
}
} catch (e) {
events.emit('error', e)
}
}
if (topic === address) {
queue.add(task)
}
}
const handlePeerDisconnected = async event => {
peers.delete(event.detail.toString())
}
/**
* Add a log entry to the Sync Protocol to be sent to peers.
* @function add
* @param {Entry} entry Log entry
* @memberof module:Sync~Sync
* @instance
*/
const add = async (entry) => {
if (started && entry && entry.hash) {
const bytes = await log.storage.get(entry.hash)
await pubsub.publish(address, bytes)
}
}
/**
* Stop the Sync Protocol.
* @function stop
* @memberof module:Sync~Sync
* @instance
*/
const stopSync = async () => {
if (started) {
started = false
await queue.clear()
pubsub.removeEventListener('subscription-change', handlePeerSubscribed)
pubsub.removeEventListener('message', handleUpdateMessage)
await libp2p.unhandle(headsSyncAddress)
await pubsub.unsubscribe(address)
libp2p.removeEventListener('peer:disconnect', handlePeerDisconnected)
peers.clear()
}
}
/**
* Start the Sync Protocol.
* @function start
* @memberof module:Sync~Sync
* @instance
*/
const startSync = async () => {
if (!started) {
pubsub.addEventListener('subscription-change', handlePeerSubscribed)
pubsub.addEventListener('message', handleUpdateMessage)
// Subscribe to the pubsub channel for this database through which updates are sent
await pubsub.subscribe(address)
// Exchange head entries with peers when connected
await libp2p.handle(headsSyncAddress, handleReceiveHeads)
// Remove disconnected peers from `peers`, as otherwise they will not resync heads on reconnection
libp2p.addEventListener('peer:disconnect', handlePeerDisconnected)
started = true
}
}
// Start Sync automatically
if (start !== false) {
await startSync()
}
return {
add,
stop: stopSync,
start: startSync,
events,
peers
}
}
export { Sync as default }
You can’t perform that action at this time.