Method syncReplicas

enterprise/replicasync/replicasync.go:235–366

(ctx context.Context)

}

func (m *Manager) syncReplicas(ctx context.Context) error {
	m.closeMutex.Lock()
	select {
	case <-m.closed:
		m.closeMutex.Unlock()
		return xerrors.New("manager is closed")
	default:
	}
	m.closeWait.Add(1)
	m.closeMutex.Unlock()
	defer m.closeWait.Done()
	// Expect replicas to update once every three times the interval...
	// If they don't, assume death!
	// nolint:gocritic // Reading replicas is a system function
	replicas, err := m.db.GetReplicasUpdatedAfter(dbauthz.AsSystemRestricted(ctx), m.updateInterval())
	if err != nil {
		return xerrors.Errorf("get replicas: %w", err)
	}

	m.mutex.Lock()
	m.peers = make([]database.Replica, 0, len(replicas))
	for _, replica := range replicas {
		if replica.ID == m.id {
			continue
		}
		// Don't peer with nodes that have an empty relay address.
		if replica.RelayAddress == "" {
			m.logger.Debug(ctx, "peer doesn't have an address, skipping",
				slog.F("replica_hostname", replica.Hostname),
			)
			continue
		}
		m.peers = append(m.peers, replica)
	}
	m.mutex.Unlock()

	client := http.Client{
		Timeout: m.options.PeerTimeout,
		Transport: &http.Transport{
			TLSClientConfig: m.options.TLSConfig,
		},
	}
	defer client.CloseIdleConnections()

	peers := m.Regional()
	errs := make(chan error, len(peers))
	for _, peer := range peers {
		go func(peer database.Replica) {
			err := PingPeerReplica(ctx, client, peer.RelayAddress)
			if err != nil {
				errs <- xerrors.Errorf("ping sibling replica %s (%s): %w", peer.Hostname, peer.RelayAddress, err)
				m.logger.Warn(ctx, "failed to ping sibling replica, this could happen if the replica has shutdown",
					slog.F("replica_hostname", peer.Hostname),
					slog.F("replica_relay_address", peer.RelayAddress),
					slog.Error(err),
				)
				return
			}