-
Notifications
You must be signed in to change notification settings - Fork 210
/
syncer.go
1163 lines (995 loc) · 31.2 KB
/
syncer.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
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package sync
import (
"encoding/hex"
"errors"
"fmt"
"github.com/spacemeshos/go-spacemesh/events"
"sync"
"time"
"github.com/spacemeshos/go-spacemesh/common/types"
"github.com/spacemeshos/go-spacemesh/database"
"github.com/spacemeshos/go-spacemesh/log"
"github.com/spacemeshos/go-spacemesh/mesh"
p2pconf "github.com/spacemeshos/go-spacemesh/p2p/config"
p2ppeers "github.com/spacemeshos/go-spacemesh/p2p/peers"
"github.com/spacemeshos/go-spacemesh/p2p/server"
"github.com/spacemeshos/go-spacemesh/p2p/service"
"github.com/spacemeshos/go-spacemesh/rand"
"github.com/spacemeshos/go-spacemesh/timesync"
"strconv"
)
type forBlockInView func(view map[types.BlockID]struct{}, layer types.LayerID, blockHandler func(block *types.Block) (bool, error)) error
type txMemPool interface {
Get(id types.TransactionID) (*types.Transaction, error)
Put(id types.TransactionID, item *types.Transaction)
}
type atxDB interface {
ProcessAtx(atx *types.ActivationTx) error
GetFullAtx(id types.ATXID) (*types.ActivationTx, error)
GetEpochAtxs(epochID types.EpochID) (atxs []types.ATXID)
}
type poetDb interface {
HasProof(proofRef []byte) bool
ValidateAndStore(proofMessage *types.PoetProofMessage) error
GetProofMessage(proofRef []byte) ([]byte, error)
}
type blockEligibilityValidator interface {
BlockSignedAndEligible(block *types.Block) (bool, error)
}
type ticker interface {
Subscribe() timesync.LayerTimer
Unsubscribe(timer timesync.LayerTimer)
GetCurrentLayer() types.LayerID
LayerToTime(types.LayerID) time.Time
}
type net struct {
peers
RequestTimeout time.Duration
*server.MessageServer
exit chan struct{}
}
func (ms net) Close() {
ms.MessageServer.Close()
ms.peers.Close()
}
func (ms net) GetTimeout() time.Duration {
return ms.RequestTimeout
}
func (ms net) GetExit() chan struct{} {
return ms.exit
}
// Configuration represents all config params needed by syncer
type Configuration struct {
LayersPerEpoch uint16
Concurrency int // number of workers for sync method
LayerSize int
RequestTimeout time.Duration
SyncInterval time.Duration
ValidationDelta time.Duration
AtxsLimit int
Hdist int
AlwaysListen bool
}
var (
errDupTx = errors.New("duplicate TransactionID in block")
errDupAtx = errors.New("duplicate ATXID in block")
errTooManyAtxs = errors.New("too many atxs in blocks")
errNoBlocksInLayer = errors.New("layer has no blocks")
errNoActiveSet = errors.New("block does not declare active set")
errZeroActiveSet = errors.New("block declares empty active set")
emptyLayer = types.Layer{}.Hash()
)
type status int
func (s *status) String() string {
if *s == 0 {
return "pending"
}
if *s == 1 {
return "inProgress"
}
return "done"
}
const (
pending status = 0
inProgress status = 1
done status = 2
blockMsg server.MessageType = 1
layerHashMsg server.MessageType = 2
layerIdsMsg server.MessageType = 3
txMsg server.MessageType = 4
atxMsg server.MessageType = 5
poetMsg server.MessageType = 6
atxIdsMsg server.MessageType = 7
atxIdrHashMsg server.MessageType = 8
syncProtocol = "/sync/1.0/"
validatingLayerNone types.LayerID = 0
)
// Syncer is used to sync the node with the network
// periodically the Syncer will check if the node is synced with the rest of the network
// and will follow the sync protocol in order to fetch all missing data and become synced again
type Syncer struct {
log.Log
Configuration
*mesh.Mesh
blockEligibilityValidator
*net
ticker
poetDb poetDb
txpool txMemPool
atxDb atxDB
validatingLayer types.LayerID
validatingLayerMutex sync.Mutex
syncLock types.TryMutex
startLock types.TryMutex
forceSync chan bool
syncTimer *time.Ticker
exit chan struct{}
gossipLock sync.RWMutex
gossipSynced status
awaitCh chan struct{}
blockQueue *blockQueue
txQueue *txQueue
atxQueue *atxQueue
}
// NewSync fires a sync every sm.SyncInterval or on force space from outside
func NewSync(srv service.Service, layers *mesh.Mesh, txpool txMemPool, atxDB atxDB, bv blockEligibilityValidator, poetdb poetDb, conf Configuration, clock ticker, logger log.Log) *Syncer {
exit := make(chan struct{})
srvr := &net{
RequestTimeout: conf.RequestTimeout,
MessageServer: server.NewMsgServer(srv.(server.Service), syncProtocol, conf.RequestTimeout, make(chan service.DirectMessage, p2pconf.Values.BufferSize), logger),
peers: p2ppeers.NewPeers(srv, logger.WithName("peers")),
exit: exit,
}
s := &Syncer{
blockEligibilityValidator: bv,
Configuration: conf,
Log: logger,
Mesh: layers,
net: srvr,
ticker: clock,
syncLock: types.TryMutex{},
poetDb: poetdb,
txpool: txpool,
atxDb: atxDB,
startLock: types.TryMutex{},
forceSync: make(chan bool),
validatingLayer: validatingLayerNone,
syncTimer: time.NewTicker(conf.SyncInterval),
exit: exit,
gossipSynced: pending,
awaitCh: make(chan struct{}),
}
s.blockQueue = newValidationQueue(srvr, conf, s)
s.txQueue = newTxQueue(s)
s.atxQueue = newAtxQueue(s, s.FetchPoetProof)
srvr.RegisterBytesMsgHandler(layerHashMsg, newLayerHashRequestHandler(layers, logger))
srvr.RegisterBytesMsgHandler(blockMsg, newBlockRequestHandler(layers, logger))
srvr.RegisterBytesMsgHandler(layerIdsMsg, newLayerBlockIdsRequestHandler(layers, logger))
srvr.RegisterBytesMsgHandler(txMsg, newTxsRequestHandler(s, logger))
srvr.RegisterBytesMsgHandler(atxMsg, newAtxsRequestHandler(s, logger))
srvr.RegisterBytesMsgHandler(poetMsg, newPoetRequestHandler(s, logger))
srvr.RegisterBytesMsgHandler(atxIdsMsg, newEpochAtxsRequestHandler(s, logger))
srvr.RegisterBytesMsgHandler(atxIdrHashMsg, newAtxHashRequestHandler(s, logger))
return s
}
// ForceSync signals syncer to run the synchronise flow
func (s *Syncer) ForceSync() {
s.forceSync <- true
}
// Close closes all running goroutines
func (s *Syncer) Close() {
s.Info("Closing syncer")
close(s.exit)
close(s.forceSync)
s.peers.Close()
s.blockQueue.Close()
s.atxQueue.Close()
s.txQueue.Close()
s.MessageServer.Close()
s.syncLock.Lock()
s.syncLock.Unlock()
s.Info("sync Closed")
}
// check if syncer was closed
func (s *Syncer) shutdown() bool {
select {
case <-s.exit:
s.Info("receive interrupt")
return true
default:
return false
}
}
// equivalent to s.LatestLayer() >= s.lastTickedLayer()-1
// means we have data from the previous layer
func (s *Syncer) weaklySynced(layer types.LayerID) bool {
return s.LatestLayer()+1 >= layer
}
func (s *Syncer) getGossipBufferingStatus() status {
s.gossipLock.RLock()
b := s.gossipSynced
s.gossipLock.RUnlock()
return b
}
// Await returns a channel that blocks until the node is synced
func (s *Syncer) Await() chan struct{} {
return s.awaitCh
}
func (s *Syncer) notifySubscribers(prevStatus, status status) {
if (status == done) == (prevStatus == done) {
return
}
if status == done {
close(s.awaitCh)
} else {
s.awaitCh = make(chan struct{})
}
}
// ListenToGossip enables other modules to check if they should listen to gossip
func (s *Syncer) ListenToGossip() bool {
return s.AlwaysListen || s.getGossipBufferingStatus() != pending
}
func (s *Syncer) setGossipBufferingStatus(status status) {
s.gossipLock.Lock()
defer s.gossipLock.Unlock()
if status == s.gossipSynced {
return
}
s.Info("setting gossip to '%s' ", status.String())
s.notifySubscribers(s.gossipSynced, status)
s.gossipSynced = status
}
// IsSynced returns true if the node is synced false otherwise
func (s *Syncer) IsSynced() bool {
s.Info("sync state w: %v, g:%v layer : %v latest: %v", s.weaklySynced(s.GetCurrentLayer()), s.getGossipBufferingStatus(), s.GetCurrentLayer(), s.LatestLayer())
return s.weaklySynced(s.GetCurrentLayer()) && s.getGossipBufferingStatus() == done
}
// Start starts the main pooling routine that checks the sync status every set interval
// and calls synchronise if the node is out of sync
func (s *Syncer) Start() {
if s.startLock.TryLock() {
s.Info("start syncer")
go s.run()
s.forceSync <- true
return
}
s.Info("syncer already started")
}
// fires a sync every sm.SyncInterval or on force sync from outside
func (s *Syncer) run() {
s.Debug("Start running")
for {
select {
case <-s.exit:
s.Debug("Work stopped")
return
case <-s.forceSync:
go s.synchronise()
case <-s.syncTimer.C:
go s.synchronise()
}
}
}
func (s *Syncer) synchronise() {
// only one concurrent synchronise
if s.syncLock.TryLock() == false {
return
}
// release synchronise lock
defer s.syncLock.Unlock()
curr := s.GetCurrentLayer()
// node is synced and blocks from current layer have already been validated
if curr == s.ProcessedLayer() {
s.Debug("node is synced")
// fully-synced, make sure we listen to p2p
s.setGossipBufferingStatus(done)
return
}
// we have all the data of the prev layers so we can simply validate
if s.weaklySynced(curr) {
s.handleWeaklySynced()
} else {
s.handleNotSynced(s.ProcessedLayer() + 1)
}
s.syncAtxs(curr)
}
func (s *Syncer) handleWeaklySynced() {
s.With().Info("Node is weakly synced",
s.LatestLayer(),
s.GetCurrentLayer())
events.ReportNodeStatusUpdate()
// handle all layers from processed+1 to current -1
s.handleLayersTillCurrent()
if s.shutdown() {
return
}
// validate current layer if more than s.ValidationDelta has passed
if err := s.handleCurrentLayer(); err != nil {
s.With().Error("Node is out of sync", log.Err(err))
s.setGossipBufferingStatus(pending)
return
}
if s.shutdown() {
return
}
// fully-synced, make sure we listen to p2p
s.setGossipBufferingStatus(done)
s.With().Info("Node is synced")
return
}
// validate all layers except current one
func (s *Syncer) handleLayersTillCurrent() {
// dont handle current
if s.ProcessedLayer()+1 >= s.GetCurrentLayer() {
return
}
s.Info("handle layers %d to %d", s.ProcessedLayer()+1, s.GetCurrentLayer()-1)
for currentSyncLayer := s.ProcessedLayer() + 1; currentSyncLayer < s.GetCurrentLayer(); currentSyncLayer++ {
if s.shutdown() {
return
}
if err := s.getAndValidateLayer(currentSyncLayer); err != nil {
if currentSyncLayer.GetEpoch().IsGenesis() {
log.Warning("failed getting layer even though we are weakly-synced currentLayer=%v lastTicked=%v err=%v ", currentSyncLayer, s.GetCurrentLayer(), err)
} else {
s.Panic("failed getting layer even though we are weakly-synced currentLayer=%v lastTicked=%v err=%v ", currentSyncLayer, s.GetCurrentLayer(), err)
}
}
}
return
}
// handle the current consensus layer if its is older than s.ValidationDelta
func (s *Syncer) handleCurrentLayer() error {
curr := s.GetCurrentLayer()
if s.LatestLayer() == curr && time.Now().Sub(s.LayerToTime(s.LatestLayer())) > s.ValidationDelta {
if err := s.getAndValidateLayer(s.LatestLayer()); err != nil {
if err != database.ErrNotFound {
s.Panic("failed handling current layer currentLayer=%v lastTicked=%v err=%v ", s.LatestLayer(), s.GetCurrentLayer(), err)
}
if err := s.SetZeroBlockLayer(curr); err != nil {
return err
}
}
}
if s.LatestLayer()+1 == curr && curr.GetEpoch().IsGenesis() {
_, err := s.GetLayer(s.LatestLayer())
if err == database.ErrNotFound {
err := s.SetZeroBlockLayer(s.LatestLayer())
return err
}
}
return nil
}
func (s *Syncer) handleNotSynced(currentSyncLayer types.LayerID) {
s.Info("Node is out of sync setting gossip-synced to false and starting sync")
events.ReportNodeStatusUpdate()
s.setGossipBufferingStatus(pending) // don't listen to gossip while not synced
// first, bring all the data of the prev layers
// Note: lastTicked() is not constant but updates as ticks are received
for ; currentSyncLayer < s.GetCurrentLayer(); currentSyncLayer++ {
s.With().Info("syncing layer", log.FieldNamed("current_sync_layer", currentSyncLayer),
log.FieldNamed("last_ticked_layer", s.GetCurrentLayer()))
if s.shutdown() {
return
}
lyr, err := s.getLayerFromNeighbors(currentSyncLayer)
if err != nil {
s.With().Info("could not get layer from neighbors", currentSyncLayer, log.Err(err))
return
}
if len(lyr.Blocks()) == 0 {
if err := s.SetZeroBlockLayer(currentSyncLayer); err != nil {
s.With().Error("handleNotSynced failed ", currentSyncLayer, log.Err(err))
return
}
}
s.ValidateLayer(lyr) // wait for layer validation
}
// if we are in the first epoch, we need to listen to gossip still
if currentSyncLayer.GetEpoch() < 3 {
s.setGossipBufferingStatus(done)
return
}
// wait for two ticks to ensure we are fully synced before we open gossip or validate the current layer
err := s.gossipSyncForOneFullLayer(currentSyncLayer)
if err != nil {
s.With().Error("Fatal: failed getting layer from db even though we listened to gossip", currentSyncLayer, log.Err(err))
}
}
func (s *Syncer) syncAtxs(currentSyncLayer types.LayerID) {
lastLayerOfEpoch := (currentSyncLayer.GetEpoch() + 1).FirstLayer() - 1
if currentSyncLayer == lastLayerOfEpoch {
err := s.syncEpochActivations(currentSyncLayer.GetEpoch())
if err != nil {
s.With().Error("cannot fetch epoch atxs ", currentSyncLayer, log.Err(err))
}
}
}
// Waits two ticks (while weakly-synced) in order to ensure that we listened to gossip for one full layer
// after that we are assumed to have all the data required for validation so we can validate and open gossip
// opening gossip in weakly-synced transition us to fully-synced
func (s *Syncer) gossipSyncForOneFullLayer(currentSyncLayer types.LayerID) error {
// listen to gossip
s.setGossipBufferingStatus(inProgress)
// subscribe and wait for two ticks
s.Info("waiting for two ticks while p2p is open, epoch %v", currentSyncLayer.GetEpoch())
ch := s.ticker.Subscribe()
if done := s.waitLayer(ch); done {
return fmt.Errorf("cloed while buffering first layer")
}
if done := s.waitLayer(ch); done {
return fmt.Errorf("cloed while buffering second layer ")
}
s.ticker.Unsubscribe(ch) // unsub, we won't be listening on this ch anymore
s.Info("done waiting for two ticks while listening to p2p")
// assumed to be weakly synced here
// just get the layers and validate
// get & validate first tick
if err := s.getAndValidateLayer(currentSyncLayer); err != nil {
if err != database.ErrNotFound {
return err
}
if err := s.SetZeroBlockLayer(currentSyncLayer); err != nil {
return err
}
}
// get & validate second tick
if err := s.getAndValidateLayer(currentSyncLayer + 1); err != nil {
if err != database.ErrNotFound {
return err
}
if err := s.SetZeroBlockLayer(currentSyncLayer + 1); err != nil {
return err
}
}
s.Info("done waiting for ticks and validation. setting gossip true")
// fully-synced - set gossip -synced to true
s.setGossipBufferingStatus(done)
return nil
}
func (s *Syncer) waitLayer(ch timesync.LayerTimer) bool {
select {
case <-ch:
s.Debug("waited one layer")
case <-s.exit:
s.Debug("exit while buffering")
return true
}
return false
}
func (s *Syncer) getLayerFromNeighbors(currentSyncLayer types.LayerID) (*types.Layer, error) {
if len(s.peers.GetPeers()) == 0 {
return nil, fmt.Errorf("no peers ")
}
// fetch layer hash from each peer
s.With().Info("fetch layer hash", currentSyncLayer)
m, err := s.fetchLayerHashes(currentSyncLayer)
if err != nil {
if err == errNoBlocksInLayer {
return types.NewLayer(currentSyncLayer), nil
}
return nil, err
}
if s.shutdown() {
return nil, fmt.Errorf("interupt")
}
// fetch ids for each hash
s.With().Info("fetch layer ids", currentSyncLayer)
blockIds, err := s.fetchLayerBlockIds(m, currentSyncLayer)
if err != nil {
return nil, err
}
if s.shutdown() {
return nil, fmt.Errorf("interupt")
}
blocksArr, err := s.syncLayer(currentSyncLayer, blockIds)
if len(blocksArr) == 0 || err != nil {
return nil, fmt.Errorf("could not get blocks for layer %v %v", currentSyncLayer, err)
}
return types.NewExistingLayer(types.LayerID(currentSyncLayer), blocksArr), nil
}
func (s *Syncer) syncEpochActivations(epoch types.EpochID) error {
s.Log.Info("syncing atxs for epoch %v", epoch)
hashes, err := s.fetchEpochAtxHashes(epoch)
if err != nil {
return err
}
atxIds, err := s.fetcEpochAtxs(hashes, epoch)
if err != nil {
return err
}
s.Info("fetched %v atxs for epoch %v actual map %v", len(atxIds), epoch, atxIds)
_, err = s.atxQueue.HandleAtxs(atxIds)
return err
}
func (s *Syncer) syncLayer(layerID types.LayerID, blockIds []types.BlockID) ([]*types.Block, error) {
ch := make(chan bool, 1)
foo := func(res bool) error {
s.Info("layer %v done", layerID)
ch <- res
return nil
}
tmr := newMilliTimer(syncLayerTime)
if res, err := s.blockQueue.addDependencies(layerID, blockIds, foo); err != nil {
return nil, fmt.Errorf("failed adding layer %v blocks to queue %v", layerID, err)
} else if res == false {
s.With().Info("no missing blocks for layer", layerID)
return s.LayerBlocks(layerID)
}
s.Info("layer %v wait for %d blocks", layerID, len(blockIds))
select {
case <-s.exit:
return nil, fmt.Errorf("received interupt")
case result := <-ch:
if !result {
return nil, fmt.Errorf("could not get all blocks for layer %v", layerID)
}
}
tmr.ObserveDuration()
return s.LayerBlocks(layerID)
}
func (s *Syncer) fastValidation(block *types.Block) error {
// block eligibility
if eligible, err := s.BlockSignedAndEligible(block); err != nil || !eligible {
return fmt.Errorf("block eligibility check failed - err %v", err)
}
// validate unique tx atx
if err := validateUniqueTxAtx(block); err != nil {
return err
}
return nil
}
func validateUniqueTxAtx(b *types.Block) error {
// check for duplicate tx id
mt := make(map[types.TransactionID]struct{}, len(b.TxIDs))
for _, tx := range b.TxIDs {
if _, exist := mt[tx]; exist {
return errDupTx
}
mt[tx] = struct{}{}
}
// check for duplicate atx id
if b.ActiveSet != nil {
ma := make(map[types.ATXID]struct{}, len(*b.ActiveSet))
for _, atx := range *b.ActiveSet {
if _, exist := ma[atx]; exist {
return errDupAtx
}
ma[atx] = struct{}{}
}
}
return nil
}
func (s *Syncer) fetchRefBlock(block *types.Block) error {
if block.RefBlock == nil {
return fmt.Errorf("called fetch ref block with nil ref block %v", block.ID())
}
_, err := s.GetBlock(*block.RefBlock)
if err != nil {
s.Log.Info("fetching block %v", *block.RefBlock)
fetched := s.fetchBlock(*block.RefBlock)
if !fetched {
return fmt.Errorf("failed to fetch ref block %v", *block.RefBlock)
}
}
return nil
}
func (s *Syncer) fetchAllReferencedAtxs(blk *types.Block) error {
var atxs []types.ATXID
//todo: patch, remove when golden atx
//if blk.ATXID != *types.EmptyATXID{
atxs = append(atxs, blk.ATXID)
//}
if blk.ActiveSet != nil {
if len(*blk.ActiveSet) > 0 {
atxs = append(atxs, *blk.ActiveSet...)
} else {
return errZeroActiveSet
}
} else {
if blk.RefBlock == nil {
return errNoActiveSet
}
}
_, err := s.atxQueue.HandleAtxs(atxs)
return err
}
func (s *Syncer) fetchBlockDataForValidation(blk *types.Block) error {
if blk.RefBlock != nil {
err := s.fetchRefBlock(blk)
if err != nil {
return err
}
}
return s.fetchAllReferencedAtxs(blk)
}
func (s *Syncer) blockSyntacticValidation(block *types.Block) ([]*types.Transaction, []*types.ActivationTx, error) {
// validate unique tx atx
if err := s.fetchBlockDataForValidation(block); err != nil {
return nil, nil, err
}
if err := s.fastValidation(block); err != nil {
return nil, nil, err
}
// data availability
txs, atxs, err := s.dataAvailability(block)
if err != nil {
return nil, nil, fmt.Errorf("DataAvailabilty failed for block %v err: %v", block.ID(), err)
}
// validate block's view
valid := s.validateBlockView(block)
if valid == false {
return nil, nil, fmt.Errorf("block %v not syntacticly valid", block.ID())
}
// validate block's votes
if valid, err := validateVotes(block, s.ForBlockInView, s.Hdist, s.Log); valid == false || err != nil {
return nil, nil, fmt.Errorf("validate votes failed for block %v, %v", block.ID(), err)
}
return txs, atxs, nil
}
func (s *Syncer) validateBlockView(blk *types.Block) bool {
ch := make(chan bool, 1)
defer close(ch)
foo := func(res bool) error {
s.With().Info("view validated",
blk.ID(),
log.Bool("result", res),
blk.LayerIndex)
ch <- res
return nil
}
if res, err := s.blockQueue.addDependencies(blk.ID(), blk.ViewEdges, foo); err != nil {
s.Error(fmt.Sprintf("block %v not syntactically valid", blk.ID()), err)
return false
} else if res == false {
s.With().Debug("no missing blocks in view",
blk.ID(),
blk.LayerIndex)
return true
}
return <-ch
}
func (s *Syncer) fetchAtx(ID types.ATXID) (*types.ActivationTx, error) {
atxs, err := s.atxQueue.HandleAtxs([]types.ATXID{ID})
if err != nil {
return nil, err
}
if len(atxs) == 0 {
return nil, fmt.Errorf("ATX %v not fetched", ID.ShortString())
}
return atxs[0], nil
}
// FetchAtxReferences fetches positioning and prev atxs from peers if they are not found in db
func (s *Syncer) FetchAtxReferences(atx *types.ActivationTx) error {
if atx.PositioningATX != *types.EmptyATXID {
s.Log.Info("going to fetch pos atx %v of atx %v", atx.PositioningATX.ShortString(), atx.ID().ShortString())
_, err := s.fetchAtx(atx.PositioningATX)
if err != nil {
return err
}
}
if atx.PrevATXID != *types.EmptyATXID {
s.Log.Info("going to fetch prev atx %v of atx %v", atx.PrevATXID.ShortString(), atx.ID().ShortString())
_, err := s.fetchAtx(atx.PrevATXID)
if err != nil {
return err
}
}
s.Log.Info("done fetching references for atx %v", atx.ID().ShortString())
return nil
}
func (s *Syncer) fetchBlock(ID types.BlockID) bool {
ch := make(chan bool, 1)
defer close(ch)
foo := func(res bool) error {
s.With().Info("single block fetched",
ID,
log.Bool("result", res))
//log.LayerID(uint64(blk.LayerIndex)))
ch <- res
return nil
}
id := types.CalcHash32(append(ID.Bytes(), []byte(strconv.Itoa(rand.Int()))...))
if res, err := s.blockQueue.addDependencies(id, []types.BlockID{ID}, foo); err != nil {
s.Error(fmt.Sprintf("block %v not syntactically valid", ID), err)
return false
} else if res == false {
// block already found
return true
}
return <-ch
}
func validateVotes(blk *types.Block, forBlockfunc forBlockInView, depth int, lg log.Log) (bool, error) {
view := map[types.BlockID]struct{}{}
for _, b := range blk.ViewEdges {
view[b] = struct{}{}
}
vote := map[types.BlockID]struct{}{}
for _, b := range blk.BlockVotes {
vote[b] = struct{}{}
}
traverse := func(b *types.Block) (stop bool, err error) {
if _, ok := vote[b.ID()]; ok {
delete(vote, b.ID())
}
return len(vote) == 0, nil
}
// traverse only through the last Hdist layers
lowestLayer := blk.LayerIndex - types.LayerID(depth)
if blk.LayerIndex < types.LayerID(depth) {
lowestLayer = 0
}
err := forBlockfunc(view, lowestLayer, traverse)
if err == nil && len(vote) > 0 {
return false, fmt.Errorf("voting on blocks out of view (or out of Hdist), %v %s", vote, err)
}
if err != nil {
return false, err
}
return true, nil
}
func (s *Syncer) dataAvailability(blk *types.Block) ([]*types.Transaction, []*types.ActivationTx, error) {
wg := sync.WaitGroup{}
wg.Add(1)
var txres []*types.Transaction
var txerr error
if len(blk.TxIDs) > 0 {
txres, txerr = s.txQueue.HandleTxs(blk.TxIDs)
}
var atxres []*types.ActivationTx
if txerr != nil {
return nil, nil, fmt.Errorf("failed fetching block %v transactions %v", blk.ID(), txerr)
}
s.With().Info("fetched all block data", blk.Fields()...)
return txres, atxres, nil
}
func (s *Syncer) fetchLayerBlockIds(m map[types.Hash32][]p2ppeers.Peer, lyr types.LayerID) ([]types.BlockID, error) {
// send request to different users according to returned hashes
idSet := make(map[types.BlockID]struct{}, s.LayerSize)
ids := make([]types.BlockID, 0, s.LayerSize)
for h, peers := range m {
NextHash:
for _, peer := range peers {
s.Debug("send request Peer: %v", peer)
ch, err := layerIdsReqFactory(lyr)(s, peer)
if err != nil {
return nil, err
}
timeout := time.After(s.Configuration.RequestTimeout)
select {
case <-s.GetExit():
s.Debug("worker received interrupt")
return nil, fmt.Errorf("interupt")
case <-timeout:
s.Error("layer ids request to %v timed out", peer)
continue
case v := <-ch:
if v != nil {
s.Debug("Peer: %v responded to layer ids request", peer)
// peer returned set with bad hash ask next peer
res := types.CalcBlocksHash32(v.([]types.BlockID), nil)
if h != res {
s.Warning("Peer: %v layer ids hash does not match request", peer)
}
for _, bid := range v.([]types.BlockID) {
if _, exists := idSet[bid]; !exists {
idSet[bid] = struct{}{}
ids = append(ids, bid)
}
}
// fetch for next hash
break NextHash
}
}
}
}
if len(ids) == 0 {
s.Info("could not get layer ids from any peer")
}
return ids, nil
}
func (s *Syncer) fetcEpochAtxs(m map[types.Hash32][]p2ppeers.Peer, epoch types.EpochID) ([]types.ATXID, error) {
// send request to different users according to returned hashes
idSet := make(map[types.ATXID]struct{}, s.LayerSize)
ids := make([]types.ATXID, 0, s.LayerSize)
for h, peers := range m {
NextHash:
for _, peer := range peers {
s.Debug("send request Peer: %v", peer)
ch, err := getEpochAtxIds(epoch, s, peer)
if err != nil {
return nil, err
}
timeout := time.After(s.Configuration.RequestTimeout)
select {
case <-s.GetExit():
s.Debug("worker received interrupt")
return nil, fmt.Errorf("interupt")
case <-timeout:
s.Error("layer ids request to %v timed out", peer)
continue
case v := <-ch:
if v != nil {
s.Debug("Peer: %v responded to epoch atx ids request", peer)
// peer returned set with bad hash ask next peer
res := types.CalcATXIdsHash32(v.([]types.ATXID), nil)
if h != res {
s.Warning("Peer: %v epoch atx ids hash does not match request", peer)
}
for _, bid := range v.([]types.ATXID) {
if _, exists := idSet[bid]; !exists {
idSet[bid] = struct{}{}
ids = append(ids, bid)
}
}
// fetch for next hash
break NextHash
}
}
}
}
if len(ids) == 0 {
s.Info("could not get atx ids from any peer")
}
return ids, nil
}
type peerHashPair struct {
peer p2ppeers.Peer
hash types.Hash32
}
func (s *Syncer) fetchLayerHashes(lyr types.LayerID) (map[types.Hash32][]p2ppeers.Peer, error) {
// get layer hash from each peer
wrk := newPeersWorker(s, s.GetPeers(), &sync.Once{}, hashReqFactory(lyr))
go wrk.Work()
m := make(map[types.Hash32][]p2ppeers.Peer)
layerHasBlocks := false
for out := range wrk.output {
pair, ok := out.(*peerHashPair)
if pair != nil && ok { // do nothing on close channel
if pair.hash != emptyLayer {
layerHasBlocks = true
m[pair.hash] = append(m[pair.hash], pair.peer)
}
}