Method runCoordinator

agent/agent.go:1906–1956

runCoordinator runs a coordinator and returns whether a reconnect should occur.

(ctx context.Context, tClient tailnetproto.DRPCTailnetClient24, network *tailnet.Conn)

// runCoordinator runs a coordinator and returns whether a reconnect
// should occur.
func (a *agent) runCoordinator(ctx context.Context, tClient tailnetproto.DRPCTailnetClient24, network *tailnet.Conn) (retErr error) {
	// we run the RPC on the hardCtx so that we have a chance to send the disconnect message if we
	// gracefully shut down.
	coordinate, err := tClient.Coordinate(a.hardCtx)
	if err != nil {
		return xerrors.Errorf("failed to connect to the coordinate endpoint: %w", err)
	}
	defer func() {
		reason, initiator := classifyCoordinatorRPCExit(ctx, retErr)
		a.logger.Debug(ctx, "disconnected from coordination RPC",
			codersdk.ConnectionDirectionServerToAgent.SlogField(),
			reason.SlogField(),
			reason.SlogExpectedField(),
			initiator.SlogField(),
			slog.Error(retErr),
		)
	}()
	defer func() {
		cErr := coordinate.Close()
		if cErr != nil {
			a.logger.Debug(ctx, "error closing Coordinate client", slog.Error(err))
		}
	}()
	a.logger.Info(ctx, "connected to coordination RPC")

	// This allows the Close() routine to wait for the coordinator to gracefully disconnect.
	disconnected := a.setCoordDisconnected()
	if disconnected == nil {
		return nil // already closed by something else
	}
	defer close(disconnected)

	ctrl := tailnet.NewAgentCoordinationController(a.logger, network)
	coordination := ctrl.New(coordinate)

	errCh := make(chan error, 1)
	go func() {
		defer close(errCh)
		select {
		case <-ctx.Done():
			err := coordination.Close(a.hardCtx)
			if err != nil {
				a.logger.Warn(ctx, "failed to close remote coordination", slog.Error(err))
			}
			return
		case err := <-coordination.Wait():
			errCh <- err
		}
	}()
	return <-errCh
}

func (a *agent) setCoordDisconnected() chan struct{} {
	a.closeMutex.Lock()