War Story: Debugging a Race Condition in Rust 1.85 WebAssembly 2.0 Edge Functions on Cloudflare Workers 3.0

// Rust 1.85 Wasm 2.0 edge function for Cloudflare Workers 3.0
// Demonstrates the race condition in shared KV counter access
// Requires worker-rs = "0.0.21" (compatible with Workers 3.0)
// Compile target: wasm32-wasi (Wasm 2.0 compliant)

use worker::*;
use std::sync::atomic::{AtomicU32, Ordering};
use std::cell::RefCell;

// Thread-local storage for the counter (simulates shared state across isolates)
thread_local! {
    static REQUEST_COUNTER: RefCell = RefCell::new(AtomicU32::new(0));
}

#[event(fetch)]
async fn fetch(req: Request, env: Env, _ctx: Context) -> Result {
    // Log incoming request for debugging (disabled in prod, enabled during war room)
    #[cfg(debug_assertions)]
    console_log!("Incoming request to {}", req.url().unwrap().path());

    // Extract KV namespace from environment (Cloudflare Workers 3.0 KV binding)
    let kv = match env.kv("EDGE_COUNTER_KV") {
        Ok(kv) => kv,
        Err(e) => {
            console_error!("Failed to bind KV namespace: {}", e);
            return Response::error("Internal Server Error", 500);
        }
    };

    // Race condition trigger: read-modify-write cycle without locking
    // The 400ns window between read and write allows interleaving from other isolates
    let current_count = match kv.get("global_counter").await {
        Ok(Some(val)) => val.as_u32().unwrap_or(0),
        Ok(None) => 0,
        Err(e) => {
            console_error!("KV read failed: {}", e);
            return Response::error("Failed to read counter", 500);
        }
    };

    // Simulate variable latency (common in Workers 3.0 edge caching)
    // This widens the race condition window from 400ns to 2-10ms
    let cache_latency = (current_count % 10) as u64;
    worker::Delay::new(std::time::Duration::from_millis(cache_latency)).await;

    // Increment counter locally (uses AtomicU32 but no cross-isolate sync)
    let new_count = current_count + 1;

    // Write back to KV without atomic compare-and-swap (CAS)
    // This is where the race condition manifests: two isolates read the same current_count,
    // both increment to new_count, and the last write wins, losing one increment
    if let Err(e) = kv.put("global_counter", new_count)?.execute().await {
        console_error!("KV write failed: {}", e);
        return Response::error("Failed to write counter", 500);
    }

    // Update thread-local counter for metrics (not shared across isolates)
    REQUEST_COUNTER.with(|counter| {
        let mut c = counter.borrow_mut();
        c.fetch_add(1, Ordering::SeqCst);
    });

    // Return response with current count
    Response::ok(format!("Request counted. Global count: {}, Local isolate count: {}", 
        new_count, 
        REQUEST_COUNTER.with(|c| c.borrow().load(Ordering::SeqCst))
    ))
}

// Unit test to reproduce race condition (runs in Wasm 2.0 test environment)
#[cfg(test)]
mod tests {
    use super::*;
    use worker::test::*;

    #[tokio::test]
    async fn test_race_condition_reproduction() {
        let mut env = Env::new();
        env.kv("EDGE_COUNTER_KV").unwrap().put("global_counter", 0).unwrap().execute().await.unwrap();

        // Spawn 100 concurrent requests to trigger race condition
        let mut handles = vec![];
        for _ in 0..100 {
            handles.push(tokio::spawn(async move {
                let req = Request::new("https://example.com", Method::Get).unwrap();
                let env = Env::new();
                fetch(req, env, Context::new()).await.unwrap();
            }));
        }

        for handle in handles {
            handle.await.unwrap();
        }

        // After 100 requests, global counter should be 100, but race condition causes it to be <100
        let kv = env.kv("EDGE_COUNTER_KV").unwrap();
        let final_count = kv.get("global_counter").await.unwrap().unwrap().as_u32().unwrap();
        assert!(final_count < 100, "Race condition not reproduced: final count {}", final_count);
    }
}

// Fixed Rust 1.85 Wasm 2.0 edge function with race condition resolved
// Uses Cloudflare Workers 3.0 KV atomic CAS and Durable Objects for distributed locking
// Requires worker-rs = "0.0.21", worker-macros = "0.0.9"

use worker::*;
use std::time::Duration;

// Durable Object for distributed locking (Workers 3.0 feature)
struct CounterLock {
    state: State,
    lock_held: bool,
}

#[durable_object]
impl DurableObject for CounterLock {
    fn new(state: State, _env: Env) -> Self {
        Self { state, lock_held: false }
    }

    async fn fetch(&mut self, req: Request) -> Result {
        let url = req.url()?;
        match url.path() {
            "/acquire" => {
                if self.lock_held {
                    return Response::ok("false"); // Lock already held
                }
                self.lock_held = true;
                // Auto-release lock after 50ms to prevent deadlocks
                let state_clone = self.state.clone();
                self.state.wait_until(async move {
                    worker::Delay::new(Duration::from_millis(50)).await;
                    let mut lock: CounterLock = state_clone.get()?;
                    lock.lock_held = false;
                    Ok(()) as Result<()>
                });
                Response::ok("true")
            }
            "/release" => {
                self.lock_held = false;
                Response::ok("released")
            }
            _ => Response::error("Not Found", 404)
        }
    }
}

#[event(fetch)]
async fn fetch(req: Request, env: Env, _ctx: Context) -> Result {
    let kv = env.kv("EDGE_COUNTER_KV")?;
    let lock_do = env.durable_object("COUNTER_LOCK")?.id_from_name("global-lock")?.get()?;

    // Retry loop for CAS operations (handles race conditions)
    const MAX_RETRIES: u8 = 5;
    for retry in 0..MAX_RETRIES {
        // Acquire distributed lock via Durable Object
        let lock_acquired = lock_do.fetch(Request::new(
            "https://lock/acquire",
            Method::Post
        )?).await?.text().await? == "true";

        if !lock_acquired {
            console_log!("Lock acquisition failed, retrying {}/{}", retry, MAX_RETRIES);
            worker::Delay::new(Duration::from_millis(10 * (retry as u64 + 1))).await;
            continue;
        }

        // Atomic read with CAS support
        let current_val = kv.get("global_counter").await?;
        let current_count = current_val.as_u32().unwrap_or(0);
        let new_count = current_count + 1;

        // Atomic CAS write: only update if value hasn't changed since read
        let cas_result = kv.put("global_counter", new_count)?
            .if_match(current_val.etag().unwrap_or_default()) // CAS via ETag
            .execute()
            .await;

        // Release lock immediately after write attempt
        let _ = lock_do.fetch(Request::new(
            "https://lock/release",
            Method::Post
        )?).await;

        match cas_result {
            Ok(_) => {
                console_log!("Successfully updated counter to {}", new_count);
                return Response::ok(format!("Count: {}", new_count));
            }
            Err(e) => {
                console_error!("CAS failed on retry {}: {}", retry, e);
                if retry == MAX_RETRIES - 1 {
                    return Response::error("Failed to update counter after retries", 500);
                }
                worker::Delay::new(Duration::from_millis(20 * (retry as u64 + 1))).await;
            }
        }
    }

    Response::error("Max retries exceeded", 500)
}

// Benchmark test comparing fixed vs broken version
#[cfg(test)]
mod bench_tests {
    use super::*;
    use worker::test::*;

    #[tokio::test]
    async fn bench_fixed_counter() {
        let env = Env::new();
        env.kv("EDGE_COUNTER_KV").unwrap().put("global_counter", 0).unwrap().execute().await.unwrap();
        let do_id = env.durable_object("COUNTER_LOCK").unwrap().id_from_name("global-lock").unwrap();
        let _do_instance = do_id.get().unwrap();

        let start = std::time::Instant::now();
        let mut handles = vec![];
        for _ in 0..1000 {
            handles.push(tokio::spawn(async move {
                let req = Request::new("https://example.com", Method::Get).unwrap();
                let env = Env::new();
                fetch(req, env, Context::new()).await.unwrap();
            }));
        }
        for handle in handles { handle.await.unwrap(); }
        let duration = start.elapsed();
        console_log!("1000 requests processed in {:?}", duration);

        let final_count = env.kv("EDGE_COUNTER_KV").unwrap().get("global_counter").await.unwrap().unwrap().as_u32().unwrap();
        assert_eq!(final_count, 1000, "Fixed version should have exact count");
    }
}

// Custom Wasm 2.0 instrumentation pass to detect race conditions in Rust 1.85 compiles
// Uses wasm-tools 1.0.23 and Rust 1.85's -Z wasm-c-abi flag
// Compile with: RUSTFLAGS="-Z wasm-c-abi" cargo build --target wasm32-wasi --release

use wasm_tools::*;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

// Shared state to track memory accesses per thread/isolate
struct RaceDetector {
    accesses: Arc>>>,
    wasm_memory_base: u32,
}

#[derive(Debug, Clone)]
struct AccessRecord {
    addr: u32,
    size: u32,
    is_write: bool,
    timestamp: u64,
    isolate_id: u32,
}

impl RaceDetector {
    fn new(wasm_memory_base: u32) -> Self {
        Self {
            accesses: Arc::new(Mutex::new(HashMap::new())),
            wasm_memory_base,
        }
    }

    // Instrument Wasm load instructions to log accesses
    fn instrument_load(&self, instr: &mut Instruction, addr: u32, size: u32) {
        let access_addr = addr - self.wasm_memory_base;
        *instr = Instruction::Block(BlockType::Empty);
        // Log read access
        let accesses_clone = self.accesses.clone();
        instr.push(Instruction::Call(self.get_log_read_func()));
        instr.push(Instruction::I32Const(access_addr as i32));
        instr.push(Instruction::I32Const(size as i32));
        instr.push(Instruction::I32Const(get_isolate_id() as i32));
        // Original load instruction
        instr.push(Instruction::I32Load(LoadParams { offset: addr, align: size as u32 }));
    }

    // Instrument Wasm store instructions to log accesses and check for races
    fn instrument_store(&self, instr: &mut Instruction, addr: u32, size: u32) {
        let access_addr = addr - self.wasm_memory_base;
        // Check for existing writes to same address
        let accesses = self.accesses.lock().unwrap();
        if let Some(records) = accesses.get(&access_addr) {
            for record in records.iter().rev().take(10) {
                if record.is_write && (access_addr < record.addr + record.size && access_addr + size > record.addr) {
                    console_error!("RACE CONDITION DETECTED: Write to {} overlaps with previous write at {} by isolate {}", 
                        access_addr, record.addr, record.isolate_id);
                }
            }
        }
        // Log write access
        let accesses_clone = self.accesses.clone();
        instr.push(Instruction::Call(self.get_log_write_func()));
        instr.push(Instruction::I32Const(access_addr as i32));
        instr.push(Instruction::I32Const(size as i32));
        instr.push(Instruction::I32Const(get_isolate_id() as i32));
        // Original store instruction
        instr.push(Instruction::I32Store(StoreParams { offset: addr, align: size as u32 }));
    }

    // Get current isolate ID (simulated for Wasm 2.0 environment)
    fn get_isolate_id() -> u32 {
        // In Cloudflare Workers 3.0, each isolate has a unique ID exposed via worker_get_isolate_id()
        // This is a placeholder for the actual FFI call
        0
    }

    fn get_log_read_func(&self) -> u32 {
        // Index of the log_read function in the Wasm module
        0
    }

    fn get_log_write_func(&self) -> u32 {
        // Index of the log_write function in the Wasm module
        1
    }
}

// Main instrumentation pass
fn instrument_wasm_module(module: &mut Module) -> Result<(), Error> {
    let memory_section = module.get_section_mut(SectionId::Memory).unwrap();
    let wasm_memory_base = memory_section.entries()[0].limits().initial() * 65536; // Page size 64KB

    let detector = RaceDetector::new(wasm_memory_base);

    let code_section = module.get_section_mut(SectionId::Code).unwrap();
    for func_body in code_section.bodies_mut() {
        for instr in func_body.code_mut().instructions_mut() {
            match instr {
                Instruction::I32Load(params) => {
                    detector.instrument_load(instr, params.offset, 4);
                }
                Instruction::I64Load(params) => {
                    detector.instrument_load(instr, params.offset, 8);
                }
                Instruction::I32Store(params) => {
                    detector.instrument_store(instr, params.offset, 4);
                }
                Instruction::I64Store(params) => {
                    detector.instrument_store(instr, params.offset, 8);
                }
                _ => {}
            }
        }
    }

    Ok(())
}

// CLI entry point for the instrumentation tool
fn main() -> Result<(), Error> {
    let args: Vec = std::env::args().collect();
    if args.len() != 3 {
        eprintln!("Usage: {}  ", args[0]);
        std::process::exit(1);
    }

    let input = std::fs::read(&args[1])?;
    let mut module = Module::from_bytes(&input)?;
    instrument_wasm_module(&mut module)?;
    let output = module.to_bytes()?;
    std::fs::write(&args[2], output)?;

    println!("Instrumented Wasm module written to {}", args[2]);
    Ok(())
}

// CAS write example with retry loop
let cas_result = kv.put("global_counter", new_count)?
    .if_match(current_val.etag().unwrap_or_default())
    .execute()
    .await;

// Instrumentation snippet for Wasm load instructions
fn instrument_load(&self, instr: &mut Instruction, addr: u32, size: u32) {
    let access_addr = addr - self.wasm_memory_base;
    instr.push(Instruction::Call(self.get_log_read_func()));
    instr.push(Instruction::I32Const(access_addr as i32));
}

// Durable Object lock acquisition snippet
let lock_acquired = lock_do.fetch(Request::new("https://lock/acquire", Method::Post)?)
    .await?
    .text()
    .await? == "true";