Client-Server и REST — Детерминированные диалоги в асинхронном мире

MODULE PathPlanningService
VARIABLES requests, responses, server_state

TypeInvariant == 
    /\ requests \in SUBSET(Request)
    /\ responses \in [Client -> SUBSET(Response)]
    /\ server_state \in {"idle", "processing"}

HandleRequest(req) ==
    /\ server_state = "idle"
    /\ server_state' = "processing"
    /\ responses' = [responses EXCEPT ![req.client] = {}]
    /\ UNCHANGED requests

ComputePath(req) ==
    /\ server_state = "processing"
    /\ \E path \in PossiblePaths(req.start, req.goal):
        responses' = [responses EXCEPT ![req.client] = {Success(path)}]
    /\ server_state' = "idle"
    /\ UNCHANGED requests

// Типобезопасный RPC на Rust с проверкой во время компиляции
#[derive(Serialize, Deserialize)]
enum ServiceRequest {
    GetSensorData(SensorId),
    SetActuator { id: ActuatorId, value: f32 },
    ComputeTransform(TransformRequest),
}

#[derive(Serialize, Deserialize)]
enum ServiceResponse {
    SensorData { id: SensorId, value: f32, timestamp: u64 },
    ActuatorSet { success: bool },
    TransformResult(Matrix4<f32>),
    Error(ServiceError),
}

struct EmbeddedRPCServer {
    services: HashMap<ServiceId, Box<dyn Service>>,
    
    fn handle_request(&self, req: ServiceRequest) -> ServiceResponse {
        match req {
            ServiceRequest::GetSensorData(id) => {
                let value = self.sensors.get(&id).read();
                ServiceResponse::SensorData { id, value, timestamp: now() }
            }
            ServiceRequest::SetActuator { id, value } => {
                let success = self.actuators.get(&id).set(value);
                ServiceResponse::ActuatorSet { success }
            }
            ServiceRequest::ComputeTransform(req) => {
                // Вычисление с гарантиями времени выполнения
                let result = self.compute_with_deadline(req, Deadline::from_millis(10));
                ServiceResponse::TransformResult(result)
            }
        }
    }
}

// Макрос для автоматической генерации клиентского кода
#[rpc_service]
trait RobotServices {
    #[timeout_ms = 50]
    fn get_sensor_data(id: SensorId) -> Result<f32, ServiceError>;
    
    #[timeout_ms = 100]
    fn set_actuator(id: ActuatorId, value: f32) -> Result<bool, ServiceError>;
}

class AdvancedPlanningService : public rclcpp::Node {
public:
    AdvancedPlanningService() : Node("planning_service") {
        // Сервис с поддержкой отмены и прогресса
        service_ = create_service<PlanPath>(
            "plan_path",
            [this](const std::shared_ptr<PlanPath::Request> request,
                   const std::shared_ptr<PlanPath::Response> response) {
                this->handle_plan_request(request, response);
            },
            rmw_qos_profile_services_default,
            planning_callback_group_  // Выделенная группа для неблокирующей обработки
        );
        
        // Сервис для потоковых ответов (частичные результаты)
        stream_service_ = create_service<StreamPlan>(
            "stream_plan",
            [this](const std::shared_ptr<StreamPlan::Request> request,
                   const std::shared_ptr<StreamPlan::Response> response) {
                this->handle_stream_request(request, response);
            }
        );
    }
    
private:
    void handle_plan_request(const PlanPath::Request::SharedPtr req,
                             PlanPath::Response::SharedPtr res) {
        // Асинхронная обработка с поддержкой отмены
        auto future = std::async(std::launch::async, [this, req]() {
            return this->compute_path_async(req);
        });
        
        // Ожидание с таймаутом
        auto status = future.wait_for(std::chrono::milliseconds(req->timeout_ms));
        if (status == std::future_status::ready) {
            res->path = future.get();
            res->success = true;
        } else {
            res->success = false;
            res->error = "Timeout exceeded";
        }
    }
    
    void handle_stream_request(const StreamPlan::Request::SharedPtr req,
                               StreamPlan::Response::SharedPtr res) {
        // Публикация промежуточных результатов через отдельный publisher
        auto progress_pub = create_publisher<PlanningProgress>(
            "/planning/progress/" + req->request_id, 10);
        
        // Итеративное планирование с обратной связью
        auto planner = IterativePlanner();
        for (const auto& partial_path : planner.compute_iteratively(req)) {
            // Отправка частичного результата
            PlanningProgress progress;
            progress.request_id = req->request_id;
            progress.path = partial_path;
            progress_pub->publish(progress);
            
            // Проверка отмены
            if (check_cancellation(req->request_id)) {
                break;
            }
        }
        
        // Финальный ответ
        res->final_path = planner.get_final_path();
    }
};

# Сервис на FastAPI с гипермедиа-ссылками
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI(title="Robot Control API", version="2026.1")

class Link(BaseModel):
    rel: str  # отношение: self, start, stop, cancel, status
    href: str
    method: str = "GET"
    timeout_ms: Optional[int] = None

class RobotState(BaseModel):
    battery: float
    position: List[float]
    velocity: List[float]
    status: str  # idle, moving, charging, error
    links: List[Link]  # Доступные действия в текущем состоянии
    
class Task(BaseModel):
    id: str
    type: str
    parameters: dict
    status: str  # pending, running, completed, failed
    progress: Optional[float] = None
    created_at: datetime
    links: List[Link]

@app.get("/robot", response_model=RobotState)
async def get_robot_state():
    """Получение состояния робота с discoverable действиями"""
    state = get_current_state()
    
    # Динамическое определение доступных действий
    links = [
        Link(rel="self", href="/robot", method="GET"),
        Link(rel="monitor", href="/robot/telemetry", method="GET"),
    ]
    
    if state.status == "idle":
        links.append(Link(rel="navigate", href="/tasks/navigate", method="POST"))
        links.append(Link(rel="scan", href="/tasks/scan", method="POST"))
    elif state.status == "moving":
        links.append(Link(rel="stop", href="/tasks/{id}/stop", method="POST"))
        links.append(Link(rel="pause", href="/tasks/{id}/pause", method="POST"))
    
    if state.battery < 20:
        links.append(Link(rel="charge", href="/tasks/charge", method="POST"))
    
    state.links = links
    return state

@app.post("/tasks/navigate", response_model=Task, status_code=202)
async def create_navigation_task(destination: List[float]):
    """Создание задачи навигации"""
    task_id = str(uuid.uuid4())
    task = Task(
        id=task_id,
        type="navigate",
        parameters={"destination": destination},
        status="pending",
        created_at=datetime.utcnow(),
        links=[
            Link(rel="self", href=f"/tasks/{task_id}", method="GET"),
            Link(rel="cancel", href=f"/tasks/{task_id}/cancel", method="POST"),
            Link(rel="progress", href=f"/tasks/{task_id}/progress", method="GET"),
        ]
    )
    
    # Асинхронный запуск задачи
    background_tasks.add_task(execute_navigation, task_id, destination)
    
    return task

@app.get("/tasks/{task_id}/progress")
async def get_task_progress(task_id: str):
    """Получение прогресса задачи через Server-Sent Events (SSE)"""
    async def event_generator():
        while True:
            progress = get_task_progress_internal(task_id)
            
            # Формат SSE
            yield f"data: {progress}\n\n"
            
            if progress.status in ["completed", "failed", "cancelled"]:
                yield f"event: done\ndata: {progress}\n\n"
                break
                
            await asyncio.sleep(0.1)  # Обновление каждые 100 мс
    
    return EventSourceResponse(event_generator())

class AdaptiveTimeoutSystem:
    def __init__(self):
        self.history = defaultdict(list)  # service -> [response_times]
        self.timeout_multiplier = 3.0  # Множитель для таймаута
    
    def get_timeout_for_service(self, service_name, default=1000):
        if service_name not in self.history or len(self.history[service_name]) < 10:
            return default
        
        times = self.history[service_name]
        
        # Использование перцентилей вместо среднего (робастность к выбросам)
        p95 = np.percentile(times, 95)
        p99 = np.percentile(times, 99)
        
        # Адаптивный таймаут: p99 + запас на вариативность
        estimated_timeout = p99 * self.timeout_multiplier
        
        # Ограничение разумными пределами
        return min(max(estimated_timeout, p95 * 2), default * 10)
    
    def record_response_time(self, service_name, response_time):
        self.history[service_name].append(response_time)
        
        # Скользящее окно последних 1000 записов
        if len(self.history[service_name]) > 1000:
            self.history[service_name] = self.history[service_name][-1000:]
        
        # Адаптация множителя на основе успешности
        success_rate = self.calculate_success_rate(service_name)
        if success_rate > 0.95:
            # Слишком консервативно — уменьшаем множитель
            self.timeout_multiplier *= 0.99
        elif success_rate < 0.8:
            # Слишком агрессивно — увеличиваем множитель
            self.timeout_multiplier *= 1.01
        
        self.timeout_multiplier = np.clip(self.timeout_multiplier, 1.5, 10.0)

class SemanticCacheProxy {
    struct CacheEntry {
        Request request;
        Response response;
        time_point expires_at;
        vector<string> semantic_tags;  // Семантические метки для инвалидации
        float confidence;  // Уверенность в актуальности
    };
    
    LRUCache<string, CacheEntry> cache;
    
    optional<Response> handle_request(const Request& req) {
        // 1. Поиск точного совпадения
        auto exact_key = hash_request(req);
        if (cache.contains(exact_key)) {
            auto& entry = cache.get(exact_key);
            if (entry.expires_at > now()) {
                return entry.response;
            }
        }
        
        // 2. Семантический поиск (похожие запросы)
        auto similar = find_similar_requests(req);
        if (!similar.empty()) {
            // Использование наиболее похожего результата с поправкой
            auto& best_match = select_best_match(similar, req);
            if (best_match.confidence > 0.9) {
                auto adapted = adapt_response(best_match.response, req);
                // Асинхронная проверка актуальности
                verify_async(req, adapted);
                return adapted;
            }
        }
        
        // 3. Прямой вызов сервиса
        auto response = call_service(req);
        
        // 4. Кэширование с семантическими метками
        CacheEntry entry{
            .request = req,
            .response = response,
            .expires_at = now() + calculate_ttl(req),
            .semantic_tags = extract_semantic_tags(req),
            .confidence = 1.0
        };
        cache.set(exact_key, entry);
        
        return response;
    }
    
    vector<string> extract_semantic_tags(const Request& req) {
        // Извлечение смысла запроса для группировки
        // Например: "navigation", "from_office", "to_lab", "avoid_dynamic"
        vector<string> tags;
        
        if (req.type == "navigate") {
            tags.push_back("navigation");
            tags.push_back("from_" + hash_position(req.start));
            tags.push_back("to_" + hash_position(req.goal));
            
            if (req.avoid_dynamic) {
                tags.push_back("avoid_dynamic");
            }
        }
        
        return tags;
    }
};

class SwarmService:
    def __init__(self, robot_id, neighbors):
        self.robot_id = robot_id
        self.neighbors = neighbors
        self.service_registry = {}  # service -> [providers]
        
        # Gossip-протокол для распространения информации о сервисах
        self.gossip = GossipProtocol()
        
    def register_service(self, service_name, capability_fn, load=1.0):
        """Регистрация локального сервиса"""
        self.service_registry[service_name] = {
            'provider': self.robot_id,
            'capability': capability_fn,
            'load': load,
            'location': self.get_position(),
            'last_heartbeat': time.time()
        }
        
        # Анонсирование доступности соседям
        self.gossip.broadcast({
            'type': 'service_announce',
            'service': service_name,
            'provider': self.robot_id,
            'location': self.get_position(),
            'load': load
        })
    
    def call_service(self, service_name, request, strategy="nearest"):
        """Вызов сервиса с выбором оптимального провайдера"""
        # Получение списка доступных провайдеров
        providers = self.discover_providers(service_name)
        
        if not providers:
            raise ServiceUnavailableError(service_name)
        
        # Выбор стратегии маршрутизации
        if strategy == "nearest":
            provider = self.select_nearest_provider(providers)
        elif strategy == "load_balanced":
            provider = self.select_least_loaded_provider(providers)
        elif strategy == "capability_aware":
            provider = self.select_most_capable_provider(providers, request)
        else:
            provider = random.choice(providers)
        
        # Локальный вызов или удалённый RPC
        if provider == self.robot_id:
            return self.service_registry[service_name]['capability'](request)
        else:
            return self.remote_rpc(provider, service_name, request)
    
    def discover_providers(self, service_name):
        """Обнаружение провайдеров через gossip"""
        providers = []
        
        # Локальные сервисы
        if service_name in self.service_registry:
            providers.append(self.robot_id)
        
        # Сервисы, анонсированные соседями
        for neighbor in self.neighbors:
            if neighbor.has_service(service_name):
                providers.append(neighbor.id)
        
        # Сервисы из глобального реестра (через gossip)
        global_providers = self.gossip.get_service_providers(service_name)
        providers.extend(global_providers)
        
        return list(set(providers))  # Удаление дубликатов

// Определение контракта в protobuf с явным версионированием
syntax = "proto3";

package robotics.v2026;

import "google/protobuf/descriptor.proto";

// Аннотация для версионирования
extend google.protobuf.ServiceOptions {
    string api_version = 50000;
    repeated string deprecated_since = 50001;
    string sunset_date = 50002;
}

service NavigationService {
    option (api_version) = "2026.1";
    option (deprecated_since) = "2025.3";  // Если deprecated
    option (sunset_date) = "2027-01-01";   // Дата удаления
    
    rpc PlanPath(PlanPathRequest) returns (PlanPathResponse) {
        option deadline = 5000;  // 5 секунд timeout
        option idempotency_level = IDEMPOTENT;
    }
    
    // Новая версия метода с улучшенной семантикой
    rpc PlanPathV2(PlanPathRequestV2) returns (PlanPathResponseV2) {
        option (api_version) = "2026.2";
        option deadline = 10000;  // 10 секунд для более сложных запросов
    }
}

message PlanPathRequest {
    string request_id = 1;
    repeated Point waypoints = 2;
    map<string, string> constraints = 3;
    
    // Поле, помеченное для удаления в будущих версиях
    string old_field = 4 [deprecated = true];
}

---------------------------- MODULE RobotServiceVerification ----------------------------
EXTENDS Naturals, Sequences, TLC

CONSTANT MaxRequests, NumClients, Timeout

VARIABLES pending_requests, active_requests, responses, server_status

Request == [id: 1..MaxRequests, client: 1..NumClients, cmd: {"move", "stop"}]

ServiceInvariant ==
    /\ pending_requests \subseteq Request
    /\ active_requests \subseteq Request
    /\ \A r \in active_requests: server_status = "processing"
    /\ \A c \in 1..NumClients: Cardinality({r \in active_requests: r.client = c}) <= 1

NoDeadlock ==
    <>[](server_status = "idle" => pending_requests = {})

Fairness ==
    \A r \in Request: WF_server_status(ProcessRequest(r))

Liveness ==
    \A r \in Request: (r \in pending_requests) ~> (r \in DOMAIN responses)

=============================================================================

<!-- Модель сервиса с таймерами в UPPAAL -->
<nta>
    <template>
        <name>RobotService</name>
        
        <location id="idle">
            <name>Idle</name>
        </location>
        
        <location id="processing">
            <name>Processing</name>
            <label kind="invariant">t <= PROCESSING_TIMEOUT</label>
        </location>
        
        <location id="error">
            <name>Error</name>
        </location>
        
        <transition>
            <source ref="idle"/>
            <target ref="processing"/>
            <label kind="synchronisation">request?</label>
            <label kind="guard">queue.size() > 0</label>
            <label kind="assignment">t := 0, current_request := queue.dequeue()</label>
        </transition>
        
        <transition>
            <source ref="processing"/>
            <target ref="idle"/>
            <label kind="synchronisation">response!</label>
            <label kind="guard">t >= MIN_PROCESSING_TIME</label>
            <label kind="assignment">send_response(current_request)</label>
        </transition>
        
        <transition>
            <source ref="processing"/>
            <target ref="error"/>
            <label kind="guard">t >= PROCESSING_TIMEOUT</label>
        </transition>
    </template>
</nta>

class ManipulationService:
    def __init__(self):
        # Два манипулятора для резервирования
        self.arms = [Manipulator(id=f"arm_{i}") for i in range(2)]
        self.arm_status = [{"healthy": True, "load": 0.0} for _ in range(2)]
        
        # Распределённый idempotency token storage
        self.idempotency_store = RedisIdempotencyStore()
        
        # Мониторинг и трассировка
        self.tracer = OpenTelemetryTracer()
        self.metrics = PrometheusMetrics()
        
    @tracing("manipulation.pick_and_place")
    @retry(retry_on=ServiceUnavailableError, max_attempts=3)
    @timeout(2000)  # 2 секунды timeout
    async def pick_and_place(self, request: ManipulationRequest) -> ManipulationResponse:
        # Проверка идемпотентности
        if await self.idempotency_store.is_duplicate(request.idempotency_key):
            return await self.idempotency_store.get_response(request.idempotency_key)
        
        with self.tracer.start_as_current_span("select_arm"):
            # Выбор оптимального манипулятора
            arm_idx = self.select_optimal_arm(request)
            if arm_idx is None:
                self.metrics.counter("arm_selection_failures").inc()
                raise ServiceUnavailableError("No available arms")
            
            arm = self.arms[arm_idx]
            
        try:
            with self.tracer.start_as_current_span("execute_trajectory"):
                # Выполнение с мониторингом прогресса
                async with self.monitor_progress(request.task_id) as progress_cb:
                    result = await arm.execute_trajectory(
                        request.trajectory,
                        progress_callback=progress_cb,
                        timeout=request.timeout_ms
                    )
                    
                response = ManipulationResponse(
                    success=True,
                    actual_trajectory=result.trajectory,
                    execution_time=result.execution_time,
                    force_profile=result.force_profile
                )
                
                # Сохранение для идемпотентности
                await self.idempotency_store.store(
                    request.idempotency_key,
                    response,
                    ttl_seconds=3600
                )
                
                self.metrics.histogram("execution_time").observe(result.execution_time)
                return response
                
        except (ArmFaultError, TrajectoryError) as e:
            self.metrics.counter("execution_errors").inc()
            
            # Переключение на резервный манипулятор
            backup_idx = 1 - arm_idx  # Другой манипулятор
            if self.arm_status[backup_idx]["healthy"]:
                self.metrics.counter("failover_events").inc()
                return await self.pick_and_place_with_arm(request, backup_idx)
            else:
                raise ServiceUnavailableError("All arms unavailable") from e
    
    def select_optimal_arm(self, request):
        """Выбор манипулятора на основе загрузки и пригодности"""
        suitable_arms = []
        
        for i, (arm, status) in enumerate(zip(self.arms, self.arm_status)):
            if not status["healthy"]:
                continue
                
            # Проверка кинематической достижимости
            if not arm.can_reach(request.target_position):
                continue
                
            # Оценка загрузки (0-1, где 1 - полностью загружен)
            load_score = status["load"]
            
            # Оценка энергоэффективности для данной траектории
            energy_score = arm.estimate_energy(request.trajectory)
            
            # Комбинированная оценка
            total_score = (
                0.6 * (1 - load_score) +  # Предпочтение менее загруженным
                0.3 * (1 - energy_score) + # Энергоэффективность
                0.1 * random.random()      # Небольшая случайность для балансировки
            )
            
            suitable_arms.append((i, total_score))
        
        if not suitable_arms:
            return None
        
        # Выбор манипулятора с наивысшей оценкой
        return max(suitable_arms, key=lambda x: x[1])[0]

class QuantumEnhancedService:
    def __init__(self):
        self.classical_server = ClassicalService()
        self.quantum_processor = QPUConnection()
        
    async def optimize_trajectory(self, request: TrajectoryRequest):
        # Гибридный алгоритм: классический + квантовый
        
        # 1. Классическая предобработка
        with self.tracer.span("classical_preprocessing"):
            initial_solution = self.classical_server.initial_solution(request)
        
        # 2. Квантовое уточнение
        with self.tracer.span("quantum_refinement"):
            # Формулировка как QUBO задачи
            qubo = self.formulate_as_qubo(initial_solution, request.constraints)
            
            # Решение на квантовом процессоре
            quantum_result = await self.quantum_processor.solve_qubo(
                qubo,
                timeout_ms=1000
            )
            
            # Декодирование результата
            refined_solution = self.decode_quantum_result(quantum_result)
        
        # 3. Классическая постобработка
        with self.tracer.span("classical_postprocessing"):
            final_solution = self.classical_server.validate_and_repair(refined_solution)
        
        return final_solution

class ExplainableAIService:
    def __init__(self):
        self.model = load_ml_model("manipulation_policy_2026")
        self.explainer = SHAPExplainer()
        
    async def decide_grasp_pose(self, request: GraspRequest):
        # Предсказание с помощью нейросети
        prediction = self.model.predict(request.scene_pointcloud)
        
        # Генерация объяснения
        explanation = self.explainer.explain(
            model=self.model,
            input_data=request.scene_pointcloud,
            prediction=prediction
        )
        
        # Проверка уверенности и fallback на классический алгоритм
        if prediction.confidence < 0.7:
            self.metrics.counter("low_confidence_fallbacks").inc()
            classical_result = self.classical_grasp_planner(request)
            
            # Возврат обоих результатов с объяснением
            return GraspResponse(
                poses=[prediction.best_pose, classical_result],
                confidence=prediction.confidence,
                explanation=explanation,
                method_used="hybrid_fallback" if prediction.confidence < 0.7 else "neural"
            )
        
        return GraspResponse(
            poses=[prediction.best_pose],
            confidence=prediction.confidence,
            explanation=explanation,
            method_used="neural"
        )