()
| 561 | } |
| 562 | |
| 563 | func (c *Chain) syncHead() (err error) { |
| 564 | // Try to fetch if the block of the current turn is not advised yet |
| 565 | h := c.rt.getNextTurn() - 1 |
| 566 | if c.rt.getHead().Height >= h { |
| 567 | return |
| 568 | } |
| 569 | |
| 570 | var ( |
| 571 | peers = c.rt.getPeers() |
| 572 | l = len(peers.Servers) |
| 573 | le = c.logEntryWithHeadState() |
| 574 | |
| 575 | child, cancel = context.WithTimeout(c.rt.ctx, c.rt.tick) |
| 576 | wg = &sync.WaitGroup{} |
| 577 | |
| 578 | totalCount, succCount, initiatingCount uint32 |
| 579 | ) |
| 580 | defer func() { |
| 581 | wg.Wait() |
| 582 | cancel() |
| 583 | |
| 584 | if totalCount > 0 && succCount == 0 { |
| 585 | if initiatingCount == totalCount { |
| 586 | err = ErrInitiating |
| 587 | } else { |
| 588 | // Set error if all RPC calls are failed |
| 589 | err = errors.New("all remote peers are unreachable") |
| 590 | } |
| 591 | } |
| 592 | }() |
| 593 | |
| 594 | for i, s := range peers.Servers { |
| 595 | // Skip local server |
| 596 | if s == c.rt.getServer() { |
| 597 | continue |
| 598 | } |
| 599 | |
| 600 | wg.Add(1) |
| 601 | go func(i int, node proto.NodeID) { |
| 602 | defer wg.Done() |
| 603 | var ( |
| 604 | ile = le.WithFields(log.Fields{"remote": fmt.Sprintf("[%d/%d] %s", i, l, node)}) |
| 605 | req = &MuxFetchBlockReq{ |
| 606 | DatabaseID: c.databaseID, |
| 607 | FetchBlockReq: FetchBlockReq{ |
| 608 | Height: h, |
| 609 | }, |
| 610 | } |
| 611 | resp = &MuxFetchBlockResp{} |
| 612 | ) |
| 613 | |
| 614 | atomic.AddUint32(&totalCount, 1) |
| 615 | if err := c.cl.CallNodeWithContext( |
| 616 | child, node, route.SQLCFetchBlock.String(), req, resp, |
| 617 | ); err != nil { |
| 618 | if !strings.Contains(err.Error(), ErrUnknownMuxRequest.Error()) { |
| 619 | ile.WithError(err).Error("failed to fetch block from peer") |
| 620 | return |
no test coverage detected