dynamo-kv-router 1.3.0

// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

use std::collections::HashMap;

use anyhow::{Context, Result};
use serde::Deserialize;

use crate::protocols::RouterEvent;

use super::registry::{IndexerKey, WorkerRegistry};

#[derive(Deserialize)]
struct DumpEntry {
    block_size: u32,
    events: Vec<RouterEvent>,
}

pub async fn recover_from_peers(peers: &[String], registry: &WorkerRegistry) -> Result<bool> {
    let client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(10))
        .build()
        .context("failed to build HTTP client")?;

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    for peer_url in peers {
        match try_recover_from_peer(&client, peer_url, registry).await {
            Ok(()) => {
                tracing::info!(peer = %peer_url, "recovery from peer succeeded");
                return Ok(true);
            }
            Err(e) => {
                tracing::warn!(peer = %peer_url, error = %e, "recovery from peer failed, trying next");
            }
        }
    }

    Ok(false)
}

async fn try_recover_from_peer(
    client: &reqwest::Client,
    peer_url: &str,
    registry: &WorkerRegistry,
) -> Result<()> {
    let dump_url = format!("{peer_url}/dump");
    tracing::info!(url = %dump_url, "fetching dump from peer");

    let resp = client
        .get(&dump_url)
        .send()
        .await
        .context("HTTP request failed")?;

    if !resp.status().is_success() {
        anyhow::bail!("peer returned status {}", resp.status());
    }

    let dump: HashMap<String, DumpEntry> =
        resp.json().await.context("failed to parse dump response")?;

    let mut total_events = 0usize;
    for (map_key, entry) in dump {
        let (model_name, tenant_id) = map_key
            .split_once(':')
            .ok_or_else(|| anyhow::anyhow!("invalid dump key format: {map_key}"))?;

        let key = IndexerKey {
            model_name: model_name.to_string(),
            tenant_id: tenant_id.to_string(),
        };

        let indexer = registry.get_or_create_indexer(key, entry.block_size);

        for event in entry.events {
            // Use the tier-aware dispatcher so HostPinned/Disk events from the
            // peer's dump land in the matching lower-tier slot rather than the
            // device primary. The peer side retags lower-tier events in
            // `Indexer::dump_events`, so the `storage_tier` here is correct.
            indexer.apply_event_routed(event).await;
            total_events += 1;
        }
    }

    tracing::info!(total_events, "applied dump events from peer");
    Ok(())
}