feat: update Godot integration and coordinated package dependencies

Runtime/ShrinkBusSchedulers.cs

@@ -1,6 +1,7 @@
 #nullable enable
 
 using System;
+using System.Buffers;
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Threading;
@@ -17,11 +18,22 @@ namespace ShrinkEventBus
     internal interface IShrinkBusScheduler : IDisposable
     {
         bool IsOnSchedulerThread { get; }
+        bool SupportsDeferred { get; }
         bool TryPost(Action action);
+        ShrinkDeferredPostStatus TryPostDeferred(Action action);
         UniTask PostAsync(Func<UniTask> action, CancellationToken cancellationToken);
         UniTask ShutdownAsync(bool drain, TimeSpan timeout);
     }
 
+    internal enum ShrinkDeferredPostStatus
+    {
+        Accepted,
+        QueueFull,
+        SchedulerRejected,
+        Unsupported,
+        Stopped
+    }
+
     internal sealed class ShrinkQueueFullException : InvalidOperationException
     {
         public ShrinkQueueFullException(string schedulerName)
@@ -81,6 +93,7 @@ namespace ShrinkEventBus
         }
 
         public bool IsEmpty => _queue.IsEmpty;
+        public int Count => _queue.Count;
 
         public bool TryEnqueue(ShrinkSchedulerWorkItem item)
         {
@@ -192,6 +205,7 @@ namespace ShrinkEventBus
     internal sealed class ShrinkInlineScheduler : IShrinkBusScheduler
     {
         public bool IsOnSchedulerThread => true;
+        public bool SupportsDeferred => false;
 
         public bool TryPost(Action action)
         {
@@ -199,6 +213,9 @@ namespace ShrinkEventBus
             return true;
         }
 
+        public ShrinkDeferredPostStatus TryPostDeferred(Action action) =>
+            ShrinkDeferredPostStatus.Unsupported;
+
         public UniTask PostAsync(Func<UniTask> action, CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
@@ -213,36 +230,53 @@ namespace ShrinkEventBus
     {
         private readonly ShrinkSchedulerQueue _queue;
         private readonly MainThreadDispatcherContract _dispatcher;
+        private readonly ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher? _deferredDispatcher;
+        private readonly object _gate = new();
         private int _pumpScheduled;
         private int _disposed;
+        private bool _discardPending;
 
         public ShrinkMainThreadScheduler(string name, ShrinkBusOptions options)
         {
             _queue = new ShrinkSchedulerQueue(name, options);
             _dispatcher = ShrinkEventBusRuntime.MainThreadDispatcher;
+            _deferredDispatcher = _dispatcher as ShrinkSDK.Runtime.IShrinkDeferredMainThreadDispatcher;
         }
 
         public bool IsOnSchedulerThread => _dispatcher.IsMainThread;
+        public bool SupportsDeferred => _deferredDispatcher != null;
 
         public bool TryPost(Action action)
         {
-            if (Volatile.Read(ref _disposed) != 0)
-                return false;
+            if (Volatile.Read(ref _disposed) != 0) return false;
             if (IsOnSchedulerThread)
             {
                 action();
                 return true;
             }
+            return Enqueue(action) == ShrinkDeferredPostStatus.Accepted;
+        }
 
-            var item = new ShrinkSchedulerWorkItem(() =>
+        public ShrinkDeferredPostStatus TryPostDeferred(Action action)
+        {
+            if (action == null) throw new ArgumentNullException(nameof(action));
+            if (_deferredDispatcher == null) return ShrinkDeferredPostStatus.Unsupported;
+            return Enqueue(action);
+        }
+
+        private ShrinkDeferredPostStatus Enqueue(Action action)
+        {
+            lock (_gate)
             {
-                action();
-                return UniTask.CompletedTask;
-            }, awaitable: false);
-            if (!_queue.TryEnqueue(item))
-                return false;
-            SchedulePump();
-            return true;
+                if (_disposed != 0) return ShrinkDeferredPostStatus.Stopped;
+                var item = new ShrinkSchedulerWorkItem(() =>
+                {
+                    action();
+                    return UniTask.CompletedTask;
+                }, awaitable: false);
+                if (!_queue.TryEnqueue(item)) return ShrinkDeferredPostStatus.QueueFull;
+                return SchedulePump() ? ShrinkDeferredPostStatus.Accepted : ShrinkDeferredPostStatus.SchedulerRejected;
+            }
         }
 
         public async UniTask PostAsync(Func<UniTask> action, CancellationToken cancellationToken)
@@ -255,19 +289,24 @@ namespace ShrinkEventBus
                 await action();
                 return;
             }
-
             var item = new ShrinkSchedulerWorkItem(action, awaitable: true);
             await _queue.EnqueueAsync(item, cancellationToken);
-            SchedulePump();
+            lock (_gate)
+            {
+                if (_disposed != 0) _queue.DropPending();
+                else SchedulePump();
+            }
             await item.Completion.AttachExternalCancellation(cancellationToken);
         }
 
         public async UniTask ShutdownAsync(bool drain, TimeSpan timeout)
         {
-            Interlocked.Exchange(ref _disposed, 1);
-            if (!drain)
-                _queue.DropPending();
-            else
+            lock (_gate)
+            {
+                _disposed = 1;
+                if (!drain) { _discardPending = true; _queue.DropPending(); }
+            }
+            if (drain)
                 await ShrinkSchedulerShutdown.WaitUntilDrainedAsync(
                     _queue, () => Volatile.Read(ref _pumpScheduled) != 0, timeout);
             _queue.Dispose();
@@ -275,34 +314,66 @@ namespace ShrinkEventBus
 
         public void Dispose()
         {
-            if (Interlocked.Exchange(ref _disposed, 1) != 0)
-                return;
-            _queue.Dispose();
+            lock (_gate)
+            {
+                if (_disposed != 0) return;
+                _disposed = 1;
+                _discardPending = true;
+                _queue.Dispose();
+            }
         }
 
-        private void SchedulePump()
+        // All queued work shares one FIFO and capacity. Prefer the deferred dispatcher
+        // so a normal worker-thread post cannot pull deferred events into an inline pump.
+        private bool SchedulePump()
         {
-            if (Interlocked.Exchange(ref _pumpScheduled, 1) != 0)
-                return;
-            if (_dispatcher.TryPost(() => PumpAsync().Forget()))
-                return;
-            Interlocked.Exchange(ref _pumpScheduled, 0);
-            ShrinkEventDiagnostics.LogException(new InvalidOperationException(
-                "The configured main-thread dispatcher rejected an EventBus pump."));
+            if (_pumpScheduled != 0) return true;
+            _pumpScheduled = 1;
+            bool accepted;
+            try
+            {
+                accepted = _deferredDispatcher != null
+                    ? _deferredDispatcher.TryPostDeferred(() => PumpAsync().Forget())
+                    : _dispatcher.TryPost(() => PumpAsync().Forget());
+            }
+            catch
+            {
+                _pumpScheduled = 0;
+                _queue.DropPending();
+                throw;
+            }
+            if (accepted) return true;
+            _pumpScheduled = 0;
+            _queue.DropPending();
+            return false;
         }
 
         private async UniTask PumpAsync()
         {
+            ShrinkSchedulerWorkItem[] batch;
+            var count = 0;
+            lock (_gate)
+            {
+                var capacity = _queue.Count;
+                batch = ArrayPool<ShrinkSchedulerWorkItem>.Shared.Rent(Math.Max(1, capacity));
+                while (count < capacity && _queue.TryDequeue(out var item)) batch[count++] = item;
+            }
             try
             {
-                while (_queue.TryDequeue(out var item))
-                    await ShrinkSchedulerWorkItemRunner.RunAsync(item);
+                for (var i = 0; i < count; i++)
+                {
+                    if (_discardPending) batch[i].Drop(nameof(ShrinkMainThreadScheduler));
+                    else await ShrinkSchedulerWorkItemRunner.RunAsync(batch[i]);
+                }
             }
             finally
             {
-                Interlocked.Exchange(ref _pumpScheduled, 0);
-                if (!_queue.IsEmpty && Volatile.Read(ref _disposed) == 0)
-                    SchedulePump();
+                ArrayPool<ShrinkSchedulerWorkItem>.Shared.Return(batch, clearArray: true);
+                lock (_gate)
+                {
+                    _pumpScheduled = 0;
+                    if (!_queue.IsEmpty) SchedulePump();
+                }
             }
         }
     }
@@ -321,6 +392,7 @@ namespace ShrinkEventBus
         }
 
         public bool IsOnSchedulerThread => false;
+        public bool SupportsDeferred => false;
 
         public bool TryPost(Action action)
         {
@@ -337,6 +409,8 @@ namespace ShrinkEventBus
             return true;
         }
 
+        public ShrinkDeferredPostStatus TryPostDeferred(Action action) => ShrinkDeferredPostStatus.Unsupported;
+
         public async UniTask PostAsync(Func<UniTask> action, CancellationToken cancellationToken)
         {
             if (Volatile.Read(ref _disposed) != 0)
@@ -417,6 +491,7 @@ namespace ShrinkEventBus
         }
 
         public bool IsOnSchedulerThread => Thread.CurrentThread.ManagedThreadId == Volatile.Read(ref _threadId);
+        public bool SupportsDeferred => false;
 
         public bool TryPost(Action action)
         {
@@ -439,6 +514,8 @@ namespace ShrinkEventBus
             return true;
         }
 
+        public ShrinkDeferredPostStatus TryPostDeferred(Action action) => ShrinkDeferredPostStatus.Unsupported;
+
         public async UniTask PostAsync(Func<UniTask> action, CancellationToken cancellationToken)
         {
             if (Volatile.Read(ref _disposed) != 0)