Method execute_kv_transfer

nodedb/src/data/executor/handlers/kv/transfer.rs:31–200

Atomic fungible transfer: source.field -= amount, dest.field += amount. Entire read-validate-write is one Data Plane pass. No TOCTOU.

(
        &mut self,
        task: &ExecutionTask,
        params: TransferParams<'_>,
    )

    ///
    /// Entire read-validate-write is one Data Plane pass. No TOCTOU.
    pub(in crate::data::executor) fn execute_kv_transfer(
        &mut self,
        task: &ExecutionTask,
        params: TransferParams<'_>,
    ) -> Response {
        let TransferParams {
            tid,
            collection,
            source_key,
            dest_key,
            field,
            amount,
        } = params;
        debug!(core = self.core_id, %collection, %field, amount, "kv transfer");

        if self.kv_engine.is_over_budget() {
            return self.response_error(task, ErrorCode::ResourcesExhausted);
        }

        let now_ms = current_ms();

        // Step 1: Read both values atomically (same core, no interleaving).
        let source_val = self.kv_engine.get(tid, collection, source_key, now_ms);
        let dest_val = self.kv_engine.get(tid, collection, dest_key, now_ms);

        let Some(source_bytes) = source_val else {
            return self.response_error(task, ErrorCode::NotFound);
        };

        // Step 2: Extract and validate source balance.
        let source_balance = match extract_numeric_field(&source_bytes, field) {
            Some(v) => v,
            None => {
                return self.response_error(
                    task,
                    ErrorCode::TypeMismatch {
                        collection: collection.to_string(),
                        detail: format!("field '{field}' is not numeric or missing"),
                    },
                );
            }
        };

        if source_balance < amount {
            return self.response_error(
                task,
                ErrorCode::InsufficientBalance {
                    collection: collection.to_string(),
                    detail: format!("source has {source_balance}, need {amount}"),
                },
            );
        }

        let dest_bytes = dest_val.unwrap_or_default();
        let dest_balance = extract_numeric_field(&dest_bytes, field).unwrap_or(0.0);

        // Step 3: Compute new values.
        let new_source = match update_numeric_field(&source_bytes, field, source_balance - amount) {