feat(runtime): own placement projection acknowledgement

src/AcDream.Runtime/Physics/RuntimePlacementProjectionSubscription.cs

@@ -0,0 +1,119 @@
+using AcDream.Runtime.Entities;
+
+namespace AcDream.Runtime.Physics;
+
+/// <summary>
+/// Presentation-only sink for canonical Runtime SetPosition receipts.
+/// Implementations must apply an exact receipt idempotently: acknowledgement
+/// can fail after a successful projection when a re-entrant Runtime mutation
+/// revises the FIFO head, and the same immutable receipt may then be retried.
+/// </summary>
+public interface IRuntimePlacementProjectionSink
+{
+    bool TryApply(in RuntimePlacementProjectionSnapshot projection);
+}
+
+/// <summary>
+/// Shared graphical/no-window subscription which projects only the oldest
+/// canonical receipt and acknowledges it only after the host sink succeeds.
+/// Runtime remains the sole position, collision, residence, and lifetime
+/// authority; this class owns only its observer subscription and one
+/// idempotency token for a projection that succeeded before acknowledgement.
+/// </summary>
+public sealed class RuntimePlacementProjectionSubscription
+    : IRuntimePlacementObserver,
+      IDisposable
+{
+    private readonly RuntimePlacementProjectionChannel _channel;
+    private readonly Func<RuntimeGenerationToken> _generation;
+    private readonly IRuntimePlacementProjectionSink _sink;
+    private IDisposable? _subscription;
+    private RuntimePlacementProjectionToken _appliedAwaitingAcknowledgement;
+    private bool _disposed;
+
+    public RuntimePlacementProjectionSubscription(
+        GameRuntime runtime,
+        IRuntimePlacementProjectionSink sink)
+        : this(
+            runtime?.Placements
+                ?? throw new ArgumentNullException(nameof(runtime)),
+            () => runtime.Generation,
+            sink)
+    {
+    }
+
+    internal RuntimePlacementProjectionSubscription(
+        RuntimePlacementProjectionChannel channel,
+        Func<RuntimeGenerationToken> generation,
+        IRuntimePlacementProjectionSink sink)
+    {
+        _channel = channel ?? throw new ArgumentNullException(nameof(channel));
+        _generation = generation
+            ?? throw new ArgumentNullException(nameof(generation));
+        _sink = sink ?? throw new ArgumentNullException(nameof(sink));
+        _subscription = _channel.Subscribe(this);
+        _ = RetryPending();
+    }
+
+    public bool HasAppliedReceiptAwaitingAcknowledgement =>
+        _appliedAwaitingAcknowledgement.IsValid;
+
+    /// <summary>
+    /// Republishes Runtime's complete still-pending FIFO. Later receipts are
+    /// ignored until the exact oldest receipt projects and acknowledges.
+    /// </summary>
+    public bool RetryPending()
+    {
+        ObjectDisposedException.ThrowIf(_disposed, this);
+        RuntimeGenerationToken generation = _generation();
+        if (_appliedAwaitingAcknowledgement.IsValid
+            && (!_channel.TryPeek(
+                    generation,
+                    out RuntimePlacementProjectionSnapshot head)
+                || head.Token != _appliedAwaitingAcknowledgement))
+        {
+            _appliedAwaitingAcknowledgement = default;
+        }
+        return _channel.RetryPending(generation);
+    }
+
+    public void OnPlacement(in RuntimePlacementDelta delta)
+    {
+        if (_disposed
+            || !_channel.TryPeek(
+                delta.Stamp.Generation,
+                out RuntimePlacementProjectionSnapshot head)
+            || head != delta.Placement)
+        {
+            return;
+        }
+
+        RuntimePlacementProjectionToken token = head.Token;
+        if (_appliedAwaitingAcknowledgement != token)
+        {
+            if (!_sink.TryApply(in head))
+                return;
+            // A sink can synchronously tear down its host while applying a
+            // receipt. Leave that receipt pending for the replacement host;
+            // disposal is never permission to acknowledge afterward.
+            if (_disposed)
+                return;
+            _appliedAwaitingAcknowledgement = token;
+        }
+
+        if (_channel.Acknowledge(delta.Stamp.Generation, token)
+            && _appliedAwaitingAcknowledgement == token)
+        {
+            _appliedAwaitingAcknowledgement = default;
+        }
+    }
+
+    public void Dispose()
+    {
+        if (_disposed)
+            return;
+        _disposed = true;
+        Interlocked.Exchange(ref _subscription, null)?.Dispose();
+        _appliedAwaitingAcknowledgement = default;
+    }
+}