Repository: influxdata/influxdb
File: influxdb3_write/src/write_buffer/mod.rs (Rust)

Code before:
    write_buffer: Arc<WriteBufferImpl>,
    memory_threshold_bytes: usize,
    check_interval: Duration,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(check_interval);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        loop {
            interval.tick().await;
            check_mem_and_force_snapshot(&write_buffer, memory_threshold_bytes).await;
        }
    })
}

async fn check_mem_and_force_snapshot(
    write_buffer: &Arc<WriteBufferImpl>,
    memory_threshold_bytes: usize,
) {
    let current_buffer_size_bytes = write_buffer.buffer.get_total_size_bytes();
    debug!(
        current_buffer_size_bytes,
        memory_threshold_bytes, "checking buffer size and snapshotting"
    );

    if current_buffer_size_bytes >= memory_threshold_bytes {
        info!(
            current_buffer_size_bytes,
            memory_threshold_bytes, "forcing snapshot as buffer size > mem threshold"
        );

        let wal = Arc::clone(&write_buffer.wal);

        let cleanup_after_snapshot = wal.force_flush_buffer().await;

        // handle snapshot cleanup outside of the flush loop
        if let Some((snapshot_complete, snapshot_info, snapshot_permit)) = cleanup_after_snapshot {
            let snapshot_wal = Arc::clone(&wal);
            tokio::spawn(async move {
                let snapshot_details = snapshot_complete.await.expect("snapshot failed");
                assert_eq!(snapshot_info, snapshot_details);

                snapshot_wal
                    .cleanup_snapshot(snapshot_info, snapshot_permit)
                    .await;
            });
        }
    }
}

#[cfg(test)]
#[allow(clippy::await_holding_lock)]

Review comment (suggestion):
I'd have this as a `warn`. Ideally, we won't be forcing snapshots. If we are, it means there's memory pressure or configuration settings that should probably be changed.

Code after:
    write_buffer: Arc<WriteBufferImpl>,
    memory_threshold_bytes: usize,
    check_interval: Duration,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(check_interval);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        loop {
            interval.tick().await;
            check_mem_and_force_snapshot(&write_buffer, memory_threshold_bytes).await;
        }
    })
}

async fn check_mem_and_force_snapshot(
    write_buffer: &Arc<WriteBufferImpl>,
    memory_threshold_bytes: usize,
) {
    let current_buffer_size_bytes = write_buffer.buffer.get_total_size_bytes();
    debug!(
        current_buffer_size_bytes,
        memory_threshold_bytes, "checking buffer size and snapshotting"
    );

    if current_buffer_size_bytes >= memory_threshold_bytes {
        warn!(
            current_buffer_size_bytes,
            memory_threshold_bytes, "forcing snapshot as buffer size > mem threshold"
        );

        let wal = Arc::clone(&write_buffer.wal);

        let cleanup_after_snapshot = wal.force_flush_buffer().await;

        // handle snapshot cleanup outside of the flush loop
        if let Some((snapshot_complete, snapshot_info, snapshot_permit)) = cleanup_after_snapshot {
            let snapshot_wal = Arc::clone(&wal);
            tokio::spawn(async move {
                let snapshot_details = snapshot_complete.await.expect("snapshot failed");
                assert_eq!(snapshot_info, snapshot_details);

                snapshot_wal
                    .cleanup_snapshot(snapshot_info, snapshot_permit)
                    .await;
            });
        }
    }
}

#[cfg(test)]
#[allow(clippy::await_holding_lock)]