Method execute_plan_inner

nodedb/src/control/exec_receiver.rs:66–202

(&self, req: ExecuteRequest)

impl LocalPlanExecutor {
    async fn execute_plan_inner(&self, req: ExecuteRequest) -> ExecuteResponse {
        // ── 1. Deadline check ─────────────────────────────────────────────────
        if req.deadline_remaining_ms == 0 {
            return ExecuteResponse::err(TypedClusterError::DeadlineExceeded { elapsed_ms: 0 });
        }

        let deadline = Duration::from_millis(req.deadline_remaining_ms).min(Duration::from_secs(
            self.state.tuning.network.default_deadline_secs,
        ));

        // ── 2. Descriptor version validation ──────────────────────────────────
        //
        // For each (collection, version) pair the caller sent, look up the local
        // descriptor version from SystemCatalog. If any version differs, the
        // caller's plan was built against a stale schema — reject with a typed
        // error so they re-plan against fresh leases.
        let catalog_ref = self.state.credentials.catalog();
        if let Some(catalog) = catalog_ref.as_ref() {
            for entry in &req.descriptor_versions {
                match catalog.get_collection(DatabaseId::DEFAULT, req.tenant_id, &entry.collection)
                {
                    Ok(Some(stored)) => {
                        // Version 0 is the pre-B.1 sentinel; treat as 1 (same
                        // floor the drain gate uses).
                        let actual = if stored.descriptor_version == 0 {
                            1
                        } else {
                            stored.descriptor_version
                        };
                        if actual != entry.version {
                            return ExecuteResponse::err(TypedClusterError::DescriptorMismatch {
                                collection: entry.collection.clone(),
                                expected_version: entry.version,
                                actual_version: actual,
                            });
                        }
                    }
                    Ok(None) => {
                        // Collection not found locally — could be a new collection
                        // the follower saw but we haven't applied yet, or a race.
                        // Treat as DescriptorMismatch so the caller re-plans.
                        if entry.version != 0 {
                            return ExecuteResponse::err(TypedClusterError::DescriptorMismatch {
                                collection: entry.collection.clone(),
                                expected_version: entry.version,
                                actual_version: 0,
                            });
                        }
                    }
                    Err(e) => {
                        return ExecuteResponse::err(TypedClusterError::Internal {
                            code: PLAN_DECODE_FAILED,
                            message: format!("catalog lookup failed: {e}"),
                        });
                    }
                }
            }
        }