forked from Nexell-AI-Network/nexelliad
/
sync.go
executable file
·323 lines (260 loc) · 8.95 KB
/
sync.go
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
318
319
320
321
322
323
package server
import (
"fmt"
"sort"
"time"
"github.com/Nautilus-Network/nautiliad/cmd/nautiluswallet/libkaspawallet"
"github.com/Nautilus-Network/nautiliad/app/appmessage"
"github.com/pkg/errors"
)
var keyChains = []uint8{libkaspawallet.ExternalKeychain, libkaspawallet.InternalKeychain}
type walletAddressSet map[string]*walletAddress
func (was walletAddressSet) strings() []string {
addresses := make([]string, 0, len(was))
for addr := range was {
addresses = append(addresses, addr)
}
return addresses
}
func (s *server) sync() error {
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
err := s.collectRecentAddresses()
if err != nil {
return err
}
err = s.refreshExistingUTXOsWithLock()
if err != nil {
return err
}
for range ticker.C {
err = s.collectFarAddresses()
if err != nil {
return err
}
err = s.collectRecentAddresses()
if err != nil {
return err
}
err = s.refreshExistingUTXOsWithLock()
if err != nil {
return err
}
}
return nil
}
const (
numIndexesToQueryForFarAddresses = 100
numIndexesToQueryForRecentAddresses = 1000
)
// addressesToQuery scans the addresses in the given range. Because
// each cosigner in a multisig has its own unique path for generating
// addresses it goes over all the cosigners and add their addresses
// for each key chain.
func (s *server) addressesToQuery(start, end uint32) (walletAddressSet, error) {
addresses := make(walletAddressSet)
for index := start; index < end; index++ {
for cosignerIndex := uint32(0); cosignerIndex < uint32(len(s.keysFile.ExtendedPublicKeys)); cosignerIndex++ {
for _, keychain := range keyChains {
address := &walletAddress{
index: index,
cosignerIndex: cosignerIndex,
keyChain: keychain,
}
addressString, err := s.walletAddressString(address)
if err != nil {
return nil, err
}
addresses[addressString] = address
}
}
}
return addresses, nil
}
// collectFarAddresses collects numIndexesToQueryForFarAddresses addresses
// from the last point it stopped in the previous call.
func (s *server) collectFarAddresses() error {
s.lock.Lock()
defer s.lock.Unlock()
err := s.collectAddresses(s.nextSyncStartIndex, s.nextSyncStartIndex+numIndexesToQueryForFarAddresses)
if err != nil {
return err
}
s.nextSyncStartIndex += numIndexesToQueryForFarAddresses
return nil
}
func (s *server) maxUsedIndexWithLock() uint32 {
s.lock.RLock()
defer s.lock.RUnlock()
return s.maxUsedIndex()
}
func (s *server) maxUsedIndex() uint32 {
maxUsedIndex := s.keysFile.LastUsedExternalIndex()
if s.keysFile.LastUsedInternalIndex() > maxUsedIndex {
maxUsedIndex = s.keysFile.LastUsedInternalIndex()
}
return maxUsedIndex
}
// collectRecentAddresses collects addresses from used addresses until
// the address with the index of the last used address + numIndexesToQueryForRecentAddresses.
// collectRecentAddresses scans addresses in batches of numIndexesToQueryForRecentAddresses,
// and releases the lock between scans.
func (s *server) collectRecentAddresses() error {
index := uint32(0)
maxUsedIndex := uint32(0)
for ; index < maxUsedIndex+numIndexesToQueryForRecentAddresses; index += numIndexesToQueryForRecentAddresses {
err := s.collectAddressesWithLock(index, index+numIndexesToQueryForRecentAddresses)
if err != nil {
return err
}
maxUsedIndex = s.maxUsedIndexWithLock()
s.updateSyncingProgressLog(index, maxUsedIndex)
}
s.lock.Lock()
if index > s.nextSyncStartIndex {
s.nextSyncStartIndex = index
}
s.lock.Unlock()
return nil
}
func (s *server) collectAddressesWithLock(start, end uint32) error {
s.lock.Lock()
defer s.lock.Unlock()
return s.collectAddresses(start, end)
}
func (s *server) collectAddresses(start, end uint32) error {
addressSet, err := s.addressesToQuery(start, end)
if err != nil {
return err
}
getBalancesByAddressesResponse, err := s.rpcClient.GetBalancesByAddresses(addressSet.strings())
if err != nil {
return err
}
err = s.updateAddressesAndLastUsedIndexes(addressSet, getBalancesByAddressesResponse)
if err != nil {
return err
}
return nil
}
func (s *server) updateAddressesAndLastUsedIndexes(requestedAddressSet walletAddressSet,
getBalancesByAddressesResponse *appmessage.GetBalancesByAddressesResponseMessage) error {
lastUsedExternalIndex := s.keysFile.LastUsedExternalIndex()
lastUsedInternalIndex := s.keysFile.LastUsedInternalIndex()
for _, entry := range getBalancesByAddressesResponse.Entries {
walletAddress, ok := requestedAddressSet[entry.Address]
if !ok {
return errors.Errorf("Got result from address %s even though it wasn't requested", entry.Address)
}
if entry.Balance == 0 {
continue
}
s.addressSet[entry.Address] = walletAddress
if walletAddress.keyChain == libkaspawallet.ExternalKeychain {
if walletAddress.index > lastUsedExternalIndex {
lastUsedExternalIndex = walletAddress.index
}
continue
}
if walletAddress.index > lastUsedInternalIndex {
lastUsedInternalIndex = walletAddress.index
}
}
err := s.keysFile.SetLastUsedExternalIndex(lastUsedExternalIndex)
if err != nil {
return err
}
return s.keysFile.SetLastUsedInternalIndex(lastUsedInternalIndex)
}
func (s *server) refreshExistingUTXOsWithLock() error {
s.lock.Lock()
defer s.lock.Unlock()
return s.refreshUTXOs()
}
// updateUTXOSet clears the current UTXO set, and re-fills it with the given entries
func (s *server) updateUTXOSet(entries []*appmessage.UTXOsByAddressesEntry, mempoolEntries []*appmessage.MempoolEntryByAddress) error {
utxos := make([]*walletUTXO, 0, len(entries))
exclude := make(map[appmessage.RPCOutpoint]struct{})
for _, entriesByAddress := range mempoolEntries {
for _, entry := range entriesByAddress.Sending {
for _, input := range entry.Transaction.Inputs {
exclude[*input.PreviousOutpoint] = struct{}{}
}
}
}
for _, entry := range entries {
if _, ok := exclude[*entry.Outpoint]; ok {
continue
}
outpoint, err := appmessage.RPCOutpointToDomainOutpoint(entry.Outpoint)
if err != nil {
return err
}
utxoEntry, err := appmessage.RPCUTXOEntryToUTXOEntry(entry.UTXOEntry)
if err != nil {
return err
}
address, ok := s.addressSet[entry.Address]
if !ok {
return errors.Errorf("Got result from address %s even though it wasn't requested", entry.Address)
}
utxos = append(utxos, &walletUTXO{
Outpoint: outpoint,
UTXOEntry: utxoEntry,
address: address,
})
}
sort.Slice(utxos, func(i, j int) bool { return utxos[i].UTXOEntry.Amount() > utxos[j].UTXOEntry.Amount() })
s.utxosSortedByAmount = utxos
return nil
}
func (s *server) refreshUTXOs() error {
// It's important to check the mempool before calling `GetUTXOsByAddresses`:
// If we would do it the other way around an output can be spent in the mempool
// and not in consensus, and between the calls its spending transaction will be
// added to consensus and removed from the mempool, so `getUTXOsByAddressesResponse`
// will include an obsolete output.
mempoolEntriesByAddresses, err := s.rpcClient.GetMempoolEntriesByAddresses(s.addressSet.strings(), true, true)
if err != nil {
return err
}
getUTXOsByAddressesResponse, err := s.rpcClient.GetUTXOsByAddresses(s.addressSet.strings())
if err != nil {
return err
}
return s.updateUTXOSet(getUTXOsByAddressesResponse.Entries, mempoolEntriesByAddresses.Entries)
}
func (s *server) isSynced() bool {
return s.nextSyncStartIndex > s.maxUsedIndex()
}
func (s *server) formatSyncStateReport() string {
maxUsedIndex := s.maxUsedIndex()
if s.nextSyncStartIndex > maxUsedIndex {
maxUsedIndex = s.nextSyncStartIndex
}
return fmt.Sprintf("scanned %d out of %d addresses (%.2f%%)",
s.nextSyncStartIndex, maxUsedIndex, float64(s.nextSyncStartIndex)*100.0/float64(maxUsedIndex))
}
func (s *server) updateSyncingProgressLog(currProcessedAddresses, currMaxUsedAddresses uint32) {
if currMaxUsedAddresses > s.maxUsedAddressesForLog {
s.maxUsedAddressesForLog = currMaxUsedAddresses
if s.isLogFinalProgressLineShown {
log.Infof("An additional set of previously used addresses found, processing...")
s.maxProcessedAddressesForLog = 0
s.isLogFinalProgressLineShown = false
}
}
if currProcessedAddresses > s.maxProcessedAddressesForLog {
s.maxProcessedAddressesForLog = currProcessedAddresses
}
if s.maxProcessedAddressesForLog >= s.maxUsedAddressesForLog {
if !s.isLogFinalProgressLineShown {
log.Infof("Wallet is synced, ready for queries")
s.isLogFinalProgressLineShown = true
}
} else {
percentProcessed := float64(s.maxProcessedAddressesForLog) / float64(s.maxUsedAddressesForLog) * 100.0
log.Infof("%d addresses of %d processed (%.2f%%)...",
s.maxProcessedAddressesForLog, s.maxUsedAddressesForLog, percentProcessed)
}
}