Method Session

internal/server/grpc_session.go:84–163

Session handles GRPC session from a repository client.

(srv grpcapi.KopiaRepository_SessionServer)

// Session handles GRPC session from a repository client.
func (s *Server) Session(srv grpcapi.KopiaRepository_SessionServer) error {
	ctx := srv.Context()

	s.serverMutex.RLock()
	dr, ok := s.rep.(repo.DirectRepository)
	s.serverMutex.RUnlock()

	if !ok {
		return status.Errorf(codes.Unavailable, "not connected to a direct repository")
	}

	log := dr.LogManager().NewLogger("grpc-session")
	ctx = contentlog.WithParams(ctx, logparam.String("span:server-session", contentlog.RandomSpanID()))

	usernameAtHostname, err := s.authenticateGRPCSession(ctx, dr)
	if err != nil {
		return err
	}

	authz := s.authorizer.Authorize(ctx, dr, usernameAtHostname)
	if authz == nil {
		authz = auth.NoAccess()
	}

	p, ok := peer.FromContext(ctx)
	if !ok {
		return status.Errorf(codes.PermissionDenied, "peer not found in context")
	}

	userLog(ctx).Infof("starting session for user %q from %v", usernameAtHostname, p.Addr)
	defer userLog(ctx).Infof("session ended for user %q from %v", usernameAtHostname, p.Addr)

	opt, err := s.handleInitialSessionHandshake(srv, dr)
	if err != nil {
		userLog(ctx).Errorf("session handshake error: %v", err)
		return err
	}

	//nolint:wrapcheck
	return repo.DirectWriteSession(ctx, dr, opt, func(ctx context.Context, dw repo.DirectRepositoryWriter) error {
		// channel to which workers will be sending errors, only holds 1 slot and sends are non-blocking.
		lastErr := make(chan error, 1)

		for req, err := srv.Recv(); err == nil; req, err = srv.Recv() {
			// propagate any error from the goroutines
			select {
			case err := <-lastErr:
				userLog(ctx).Errorf("error handling session request: %v", err)

				contentlog.Log1(ctx, log,
					"error handling session request",
					logparam.Error("error", err))

				return err

			default:
			}