You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: server.go
    path of file: ./repos/moby/vendor/github.com/moby/swarmkit/v2/ca
    
    the code of the file until where you have to start completion: package ca

import (
	"bytes"
	"context"
	"crypto/subtle"
	"crypto/x509"
	"sync"
	"time"

	gogotypes "github.com/gogo/protobuf/types"
	"github.com/moby/swarmkit/v2/api"
	"github.com/moby/swarmkit/v2/api/equality"
	"github.com/moby/swarmkit/v2/identity"
	"github.com/moby/swarmkit/v2/log"
	"github.com/moby/swarmkit/v2/manager/state/store"
	"github.com/pkg/errors"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const (
	defaultReconciliationRetryInterval = 10 * time.Second
	defaultRootReconciliationInterval  = 3 * time.Second
)

// Server is the CA and NodeCA API gRPC server.
// TODO(aaronl): At some point we may want to have separate implementations of
// CA, NodeCA, and other hypothetical future CA services. At the moment,
// breaking it apart doesn't seem worth it.
type Server struct {
	mu                          sync.Mutex
	wg                          sync.WaitGroup
	ctx                         context.Context
	cancel                      func()
	store                       *store.MemoryStore
	securityConfig              *SecurityConfig
	clusterID                   string
	localRootCA                 *RootCA
	externalCA                  *ExternalCA
	externalCAPool              *x509.CertPool
	joinTokens                  *api.JoinTokens
	reconciliationRetryInterval time.Duration

	// pending is a map of nodes with pending certificates issuance or
	// renewal. They are indexed 
    
func (s *Server) Run(ctx context.Context) error {
	s.mu.Lock()
	if s.isRunning() {
		s.mu.Unlock()
		return errors.New("CA signer is already running")
	}
	s.wg.Add(1)
	s.ctx, s.cancel = context.WithCancel(log.WithModule(ctx, "ca"))
	ctx = s.ctx
	s.mu.Unlock()
	defer s.wg.Done()
	defer func() {
		s.mu.Lock()
		s.mu.Unlock()
	}()

	// Retrieve the channels to keep track of changes in the cluster
	// Retrieve all the currently registered nodes
	var (
		nodes   []*api.Node
		cluster *api.Cluster
		err     error
	)
	updates, cancel, err := store.ViewAndWatch(
		s.store,
		func(readTx store.ReadTx) error {
			cluster = store.GetCluster(readTx, s.clusterID)
			if cluster == nil {
				return errors.New("could not find cluster object")
			}
			nodes, err = store.FindNodes(readTx, store.All)
			return err
		},
		api.EventCreateNode{},
		api.EventUpdateNode{},
		api.EventDeleteNode{},
		api.EventUpdateCluster{
			Cluster: &api.Cluster{ID: s.clusterID},
			Checks:  []api.ClusterCheckFunc{api.ClusterCheckID},
		},
	)

	// call once to ensure that the join tokens and local/external CA signer are always set
	rootReconciler := &rootRotationReconciler{
		ctx:                 log.WithField(ctx, "method", "(*Server).rootRotationReconciler"),
		clusterID:           s.clusterID,
		store:               s.store,
		batchUpdateInterval: s.rootReconciliationRetryInterval,
	}

	s.UpdateRootCA(ctx, cluster, rootReconciler)

	// Do this after updateCluster has been called, so Ready() and isRunning never returns true without
	// the join tokens and external CA/security config's root CA being set correctly
	s.mu.Lock()
	close(s.started)
	s.mu.Unlock()

	if err != nil {
		log.G(ctx).WithFields(log.Fields{
			"method": "(*Server).Run",
		}).WithError(err).Errorf("snapshot store view failed")
		return err
	}
	defer cancel()

	// We might have missed some updates if there was a leader election,
	// so let's pick up the slack.
	if err := s.reconcileNodeCertificates(ctx, nodes); err != nil {
		// We don't return here because that means the Run loop would
		// never run. Log an error instead.
		log.G(ctx).WithFields(log.Fields{
			"method": "(*Server).Run",
		}).WithError(err).Errorf("error attempting to reconcile certificates")
	}

	ticker := time.NewTicker(s.reconciliationRetryInterval)
	defer ticker.Stop()

	externalTLSCredsChange, externalTLSWatchCancel := s.securityConfig.Watch()
	defer externalTLSWatchCancel()

	// Watch for new nodes being created, new nodes being updated, and changes
	// to the cluster
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		select {
		case event := <-updates:
			switch v := event.(type) {
			case api.EventCreateNode:
				s.evaluateAndSignNodeCert(ctx, v.Node)
				rootReconciler.UpdateNode(v.Node)
			case api.EventUpdateNode:
				// If this certificate is already at a final state
				// no need to evaluate and sign it.
				if !isFinalState(v.Node.Certificate.Status) {
					s.evaluateAndSignNodeCert(ctx, v.Node)
				}
				rootReconciler.UpdateNode(v.Node)
			case api.EventDeleteNode:
				rootReconciler.DeleteNode(v.Node)
			case api.EventUpdateCluster:
				if v.Cluster.ID == s.clusterID {
					s.UpdateRootCA(ctx, v.Cluster, rootReconciler)
				}
			}
		case <-externalTLSCredsChange:
			// The TLS certificates can rotate independently of the root CA (and hence which roots the
			// external CA trusts) and external CA URLs.  It's possible that the root CA update is received
			// before the external TLS cred change notification.  During that period, it is possible that
			// the TLS creds will expire or otherwise fail to authorize against external CAs.  However, in
			// that case signing will just fail with a recoverable connectivity error - the state of the
			// certificate issuance is left as pending, and on the next tick, the server will try to sign
			// all nodes with pending certs again (by which time the TLS cred change will have been
			// received).

			// Note that if the external CA changes, the new external CA *MUST* trust the current server's
			// certificate issuer, and this server's certificates should not be extremely close to expiry,
			// otherwise this server would not be able to get new TLS certificates and will no longer be
			// able to function.
			s.signingMu.Lock()
			s.externalCA.UpdateTLSConfig(NewExternalCATLSConfig(
				s.securityConfig.ClientTLSCreds.Config().Certificates, s.externalCAPool))
			s.signingMu.Unlock()
		case <-ticker.C:
			for _, node := range s.pending {
				if err := s.evaluateAndSignNodeCert(ctx, node); err != nil {
					// If this sign operation did not succeed, the rest are
					// unlikely to. Yield so that we don't hammer an external CA.
					// Since the map iteration order is randomized, there is no
					// risk of getting stuck on a problematic CSR.
					break
				}
			}
		case <-ctx.Done():
			return nil
		}
	}
}