#1270 Fix "ring buffer too small for writing block" error

by replacing the old ring buffer with a new one when its space is exhausted (allocation
happens in worker thread)

base/Cargo.toml

@@ -33,6 +33,8 @@ anyhow.workspace = true
 thiserror.workspace = true
 camino.workspace = true
 indexmap.workspace = true
+futures.workspace = true
+tokio.workspace = true
 
 [target.'cfg(target_os = "macos")'.dependencies]
 # For not letting device_query panic when macOS accessibility permissions not granted

base/src/lib.rs

@@ -47,3 +47,5 @@ pub mod panic_util;
 
 mod approx_f64;
 pub use approx_f64::*;
+
+pub mod replenishment_channel;

base/src/replenishment_channel.rs

@@ -0,0 +1,49 @@
+use futures::future::BoxFuture;
+use tokio::sync::mpsc::Receiver;
+use tracing::debug;
+
+/// An orchestration (task and receiver) to be used to supply the receiver with spare parts that might or might not be
+/// necessary.
+///
+/// This is usually used in scenarios where the consumer lives in a thread that is not allowed to allocate
+/// (for example, real-time threads).
+pub struct ReplenishmentOrchestration<T, F> {
+    pub task: F,
+    pub receiver: ReplenishmentReceiver<T>,
+}
+
+/// Creates an orchestration.
+///
+/// The capacity should be very low, depending on how many spare items you want to create.
+pub fn orchestrate_replenishment<T>(
+    capacity: usize,
+    mut create_next_item: impl FnMut() -> T + Send + 'static,
+) -> ReplenishmentOrchestration<T, BoxFuture<'static, ()>>
+where
+    T: Send + 'static,
+{
+    let (sender, receiver) = tokio::sync::mpsc::channel::<T>(capacity);
+    let task = async move {
+        while let Ok(permit) = sender.reserve().await {
+            debug!("Replenishment channel has capacity. Create next item.");
+            let item = create_next_item();
+            permit.send(item);
+        }
+    };
+    ReplenishmentOrchestration {
+        receiver: ReplenishmentReceiver { receiver },
+        task: Box::pin(task),
+    }
+}
+
+#[derive(Debug)]
+pub struct ReplenishmentReceiver<T> {
+    receiver: Receiver<T>,
+}
+
+impl<T> ReplenishmentReceiver<T> {
+    /// Returns the next available item if one is available.
+    pub fn request_item(&mut self) -> Option<T> {
+        self.receiver.try_recv().ok()
+    }
+}

main/src/infrastructure/plugin/backbone_shell.rs

@@ -2981,6 +2981,7 @@ mod playtime_impl {
     use base::metrics_util::{record_duration, record_occurrence};
     use base::spawn_in_main_thread;
     use camino::Utf8PathBuf;
+    use futures::future::BoxFuture;
     use playtime_api::persistence::PlaytimeSettings;
     use playtime_clip_engine::PlaytimeEngine;
     use reaper_high::{GroupingBehavior, Project, Reaper};
@@ -3061,8 +3062,8 @@ mod playtime_impl {
                 None
             };
         #[derive(Debug)]
-        struct RealearnPlaytimeIntegration;
-        impl playtime_clip_engine::PlaytimeIntegration for RealearnPlaytimeIntegration {
+        struct HelgoboxPlaytimeIntegration;
+        impl playtime_clip_engine::PlaytimeIntegration for HelgoboxPlaytimeIntegration {
             fn export_to_clipboard(
                 &self,
                 item: &dyn playtime_clip_engine::PlaytimeItem,
@@ -3086,12 +3087,16 @@ mod playtime_impl {
                 fs::write(settings_path, json)?;
                 Ok(())
             }
+
+            fn spawn_in_async_runtime(&self, f: BoxFuture<'static, ()>) {
+                BackboneShell::get().spawn_in_async_runtime(f);
+            }
         }
         let args = playtime_clip_engine::PlaytimeEngineInitArgs {
             available_licenses: license_manager.licenses(),
             settings: BackboneShell::read_playtime_settings(),
             metrics_recorder,
-            integration: Box::new(RealearnPlaytimeIntegration),
+            integration: Box::new(HelgoboxPlaytimeIntegration),
         };
         PlaytimeEngine::make_available_globally(PlaytimeEngine::new(args));
     }