Function flush_ilp_batch_inner

nodedb/src/control/server/ilp_batch.rs:88–230

Inner dispatch logic for ILP batch (separated for clean quota bookkeeping).

(
    state: &SharedState,
    tenant_id: TenantId,
    batch: &str,
)

/// Inner dispatch logic for ILP batch (separated for clean quota bookkeeping).
async fn flush_ilp_batch_inner(
    state: &SharedState,
    tenant_id: TenantId,
    batch: &str,
) -> crate::Result<u64> {
    // Fast path: extract collection from first line.
    let collection = batch
        .lines()
        .find(|l| !l.is_empty() && !l.starts_with('#'))
        .and_then(|l| l.split([',', ' ']).next())
        .unwrap_or("default_metrics")
        .to_string();

    // Route all ILP lines for a collection to the same vShard as the
    // collection-based scan uses. This ensures timeseries scans find
    // the memtable data on the correct Data Plane core.
    // Per-series sharding is deferred until the scan path supports
    // fan-out across multiple cores.
    let collection_vshard = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &collection);
    let mut shard_batches: std::collections::HashMap<u32, String> =
        std::collections::HashMap::new();

    for line in batch.lines() {
        if line.is_empty() || line.starts_with('#') {
            continue;
        }

        let entry = shard_batches.entry(collection_vshard.as_u32()).or_default();
        entry.push_str(line);
        entry.push('\n');
    }

    let mut total_accepted = 0u64;

    for (shard_id, shard_batch) in &shard_batches {
        let vshard_id = VShardId::new(*shard_id);
        let payload_bytes = shard_batch.as_bytes().to_vec();

        // Append to WAL first — returns the assigned LSN for dedup tracking.
        let wal_lsn = crate::control::server::wal_dispatch::wal_append_timeseries(
            &state.wal,
            tenant_id,
            vshard_id,
            crate::types::DatabaseId::DEFAULT,
            &collection,
            &payload_bytes,
            Some(&state.credentials),
        )?
        .map(|lsn| lsn.as_u64());

        let plan = PhysicalPlan::Timeseries(TimeseriesOp::Ingest {
            collection: collection.clone(),
            payload: payload_bytes,
            format: "ilp".to_string(),
            wal_lsn,
            surrogates: Vec::new(),
        });