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

Code before:
            // means even though there is a wal period the query buffer might
            // still be empty. The reason is, when snapshots are happening very
            // close to each other (when force snapshot is set), they could get
            // queued to run immediately one after the other as illustrated in
            // example series of flushes and force snapshots below,
            //
            //   1 (only wal flush) // triggered by flush interval 1s
            //   2 (snapshot)       // triggered by flush interval 1s
            //   3 (force_snapshot) // triggered by mem check interval 10s
            //   4 (force_snapshot) // triggered by mem check interval 10s
            //
            // Although the flush interval an mem check intervals aren't same
            // there's a good chance under high memory pressure there will be
            // a lot of overlapping.
            //
            // In this setup - after 2 (snapshot), we emptied wal buffer and as
            // soon as snapshot is done, 3 will try to run the snapshot but wal
            // buffer can be empty at this point, which means it adds a no-op.
            // no-op has the current time which will be used as the
            // end_time_marker. That would evict everything from query buffer, so
            // when 5 forces another snapshot there's no data in the query
            // buffer though it has a wal_period. Under normal (i.e without
            // force_snapshot) snapshot runs snapshot_tracker will check
            // wal_periods are empty so it won't trigger a snapshot in the first
            // place.
            if !persist_jobs_empty {
                loop {
                    debug!(?persisted_snapshot, ">>> TRYING TO PERSIST SNAPSHOT");
                    match persister.persist_snapshot(&persisted_snapshot).await {
                        Ok(_) => {
                            let persisted_snapshot = Some(persisted_snapshot.clone());
                            notify_snapshot_tx
                                .send(persisted_snapshot)
                                .expect("persisted snapshot notify tx should not be closed");
                            break;
                        }
                        Err(e) => {
                            error!(%e, "Error persisting snapshot, sleeping and retrying...");
                            tokio::time::sleep(Duration::from_secs(1)).await;
                        }
                    }
                }
            }

            // clear out the write buffer and add all the persisted files to the persisted files
            // on a background task to ensure that the cache has been populated before we clear
            // the buffer
            tokio::spawn(async move {
                // wait on the cache updates to complete if there is a cache:
                for notifier in cache_notifiers.into_iter().flatten() {
                    let _ = notifier.await;

Review comment (suggestion):
This check is only needed here to fix the issue but I've also made clearing the snapshots further down conditional (whether persist jobs were empty or not). I did wonder if I can work around the problem further up the call stack to avoid snapshotting (calling this function) in the first place but knowing whether query buffer is empty isn't straight forward. If there is a better way to achieve this without scattering checks on a flag that would be desirable.

Code after:
            // means even though there is a wal period the query buffer might
            // still be empty. The reason is, when snapshots are happening very
            // close to each other (when force snapshot is set), they could get
            // queued to run immediately one after the other as illustrated in
            // example series of flushes and force snapshots below,
            //
            //   1 (only wal flush) // triggered by flush interval 1s
            //   2 (snapshot)       // triggered by flush interval 1s
            //   3 (force_snapshot) // triggered by mem check interval 10s
            //   4 (force_snapshot) // triggered by mem check interval 10s
            //
            // Although the flush interval an mem check intervals aren't same
            // there's a good chance under high memory pressure there will be
            // a lot of overlapping.
            //
            // In this setup - after 2 (snapshot), we emptied wal buffer and as
            // soon as snapshot is done, 3 will try to run the snapshot but wal
            // buffer can be empty at this point, which means it adds a no-op.
            // no-op has the current time which will be used as the
            // end_time_marker. That would evict everything from query buffer, so
            // when 4 (force snapshot) runs there's no data in the query
            // buffer though it has a wal_period. When normal (i.e without
            // force_snapshot) snapshot runs, snapshot_tracker will check if
            // wal_periods are empty so it won't trigger a snapshot in the first
            // place.
            if !persist_jobs_empty {
                loop {
                    match persister.persist_snapshot(&persisted_snapshot).await {
                        Ok(_) => {
                            let persisted_snapshot = Some(persisted_snapshot.clone());
                            notify_snapshot_tx
                                .send(persisted_snapshot)
                                .expect("persisted snapshot notify tx should not be closed");
                            break;
                        }
                        Err(e) => {
                            error!(%e, "Error persisting snapshot, sleeping and retrying...");
                            tokio::time::sleep(Duration::from_secs(1)).await;
                        }
                    }
                }
            }

            // clear out the write buffer and add all the persisted files to the persisted files
            // on a background task to ensure that the cache has been populated before we clear
            // the buffer
            tokio::spawn(async move {
                // wait on the cache updates to complete if there is a cache:
                for notifier in cache_notifiers.into_iter().flatten() {
                    let _ = notifier.await;
                }