admin/Playa
1
0
Fork 0
forked from admin/VolcanicAgenticCore
Code Pull Requests Activity

feat(telegram): per-chat inbox debouncer with ordered leases

Browse Source 
Spec: ChatInbox takes an optional ILogger (dispose count, detached leases, flush failures).
This commit is contained in:
EugeneTes
2026-09-26 19:57:23 +02:00
parent b82dd859c6
commit 8e31dc61b9
4 changed files with 451 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
docs/superpowers/specs/2026-09-26-telegram-message-context-design.md
View File

@@ -154,7 +154,7 @@ router holds that message's lease.** A preparer never talks to `ChatInbox` itsel
```csharp
public sealed class ChatInbox(Func<IReadOnlyList<InboundPart>, Task> onFlush, TimeSpan window,
TimeProvider time, TimeSpan? maxHold = null) : IAsyncDisposable
TimeProvider time, TimeSpan? maxHold = null, ILogger? logger = null) : IAsyncDisposable
{
public InboundLease Begin(int messageId); // reserves the next slot; cancels an armed timer
public void Discard(); // /new, /resume: drop buffered and in-flight parts (generation bump)
@@ -178,7 +178,9 @@ Rules (one lock, timer from `TimeProvider.CreateTimer`):
4. When a slot has been open for `maxHold` (default 2 min, tracked by a second timer), it is detached. The batch is
handled as if that slot were empty. Its later `Complete` is treated as a new `Begin` + `Complete` at the end.
5. `Discard` bumps a generation counter and clears the slots. Leases from an older generation complete into nothing.
6. `DisposeAsync` (from `ChatContext.DisposeAsync`) discards and logs how many parts were dropped.
6. `DisposeAsync` (from `ChatContext.DisposeAsync`) discards and logs how many parts were dropped. The optional
logger also records a hung lease being detached and a failing `onFlush`, which is caught so it can't stop later
batches.
Race walk-through (comment, then a forwarded voice that takes 3 s to transcribe):

240
hosts/Volcanic.AgenticCore.Telegram/Inbound/ChatInbox.cs Normal file
View File

@@ -0,0 +1,240 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Volcanic.AgenticCore.Telegram.Inbound;
/// <summary>
/// Groups one chat's messages into batches. The router takes a lease per message, synchronously and in update order, and
/// completes it with the message's <see cref="InboundPart"/> once composed. A batch is flushed <c>window</c> after the
/// last completion, and only while no lease is open. Parts keep lease order (Telegram order), not completion order.
/// A lease open longer than <c>maxHold</c> stops holding the batch; its late part becomes a batch of its own.
/// </summary>
public sealed class ChatInbox : IAsyncDisposable
{
/// <summary>How long an open lease may hold a batch before it is detached.</summary>
public static readonly TimeSpan DefaultMaxHold = TimeSpan.FromMinutes(2);
internal sealed class Slot
{
public InboundPart? Part;
public bool Open = true;
public bool Detached;
public ITimer? Hold;
}
private readonly Func<IReadOnlyList<InboundPart>, Task> _onFlush;
private readonly TimeSpan _window;
private readonly TimeSpan _maxHold;
private readonly TimeProvider _time;
private readonly ILogger _logger;
private readonly Lock _lock = new();
private readonly List<Slot> _slots = [];
private ITimer? _timer;
private long _timerVersion;
private long _generation;
private int _flushing;
private bool _disposed;
/// <param name="onFlush">Receives each batch, in slot order. A failure is logged and doesn't stop later batches.</param>
/// <param name="window">Quiet time after the last completion before a batch is flushed.</param>
/// <param name="time">Clock and timers.</param>
/// <param name="maxHold">How long an open lease may hold the batch; <see cref="DefaultMaxHold"/> when null.</param>
/// <param name="logger">Records dropped parts on dispose, detached leases and failing flushes.</param>
public ChatInbox(Func<IReadOnlyList<InboundPart>, Task> onFlush, TimeSpan window, TimeProvider time, TimeSpan? maxHold = null, ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(onFlush);
ArgumentNullException.ThrowIfNull(time);
ArgumentOutOfRangeException.ThrowIfLessThan(window, TimeSpan.Zero);
_onFlush = onFlush;
_window = window;
_time = time;
_maxHold = maxHold ?? DefaultMaxHold;
_logger = logger ?? NullLogger.Instance;
}
/// <summary>Slots holding a part (test view).</summary>
internal int Buffered { get { lock (_lock) return _slots.Count(s => s.Part is not null); } }
/// <summary>Slots still open (test view).</summary>
internal int OpenLeases { get { lock (_lock) return _slots.Count(s => s.Open); } }
/// <summary>Reserve the next slot. Cancels an armed flush until this lease completes.</summary>
public InboundLease Begin(int messageId)
{
lock (_lock)
{
var slot = new Slot();
var lease = new InboundLease(this, slot, _generation, messageId);
if (_disposed)
{
slot.Open = false; // completes into nothing
return lease;
}
CancelTimer();
_slots.Add(slot);
slot.Hold = _time.CreateTimer(_ => Detach(lease), null, _maxHold, Timeout.InfiniteTimeSpan);
return lease;
}
}
/// <summary>Drop buffered parts and in-flight leases (<c>/new</c>, <c>/resume</c>). Leases already open complete into nothing.</summary>
public void Discard()
{
lock (_lock) DropLocked();
}
/// <summary>Completes when nothing is buffered, no lease is open and the last flush was handed over.</summary>
public async Task DrainAsync()
{
while (true)
{
lock (_lock)
{
if (_slots.Count == 0 && _flushing == 0) return;
}
await Task.Delay(10).ConfigureAwait(false);
}
}
/// <summary>Discards everything buffered and logs how many parts were dropped.</summary>
public ValueTask DisposeAsync()
{
int dropped;
lock (_lock)
{
if (_disposed) return ValueTask.CompletedTask;
_disposed = true;
dropped = DropLocked();
}
if (dropped > 0) _logger.LogInformation("Dropped {Count} buffered message(s) on shutdown", dropped);
return ValueTask.CompletedTask;
}
internal void Complete(InboundLease lease, InboundPart? part)
{
lock (_lock)
{
var slot = lease.Slot;
slot.Hold?.Dispose();
if (_disposed || lease.Generation != _generation) return; // discarded
if (slot.Detached)
{
if (part is null) return;
_slots.Add(new Slot { Part = part, Open = false }); // a late part: a new Begin + Complete at the end
}
else
{
if (!slot.Open) return;
slot.Open = false;
slot.Part = part;
}
SettleLocked();
}
}
private void Detach(InboundLease lease)
{
lock (_lock)
{
var slot = lease.Slot;
if (_disposed || lease.Generation != _generation || !slot.Open) return;
slot.Open = false;
slot.Detached = true;
_slots.Remove(slot);
_logger.LogWarning("Message {MessageId} took longer than {MaxHold} to prepare; sending the others without it", lease.MessageId, _maxHold);
SettleLocked();
}
}
/// <summary>Arms the flush timer when no slot is open and one holds a part; clears the slots when all are empty.</summary>
private void SettleLocked()
{
if (_slots.Any(s => s.Open)) return;
if (_slots.All(s => s.Part is null))
{
_slots.Clear();
return;
}
CancelTimer();
var version = _timerVersion;
_timer = _time.CreateTimer(_ => OnTimer(version), null, _window, Timeout.InfiniteTimeSpan);
}
/// <summary>Bumps the version so a callback from a cancelled arm can't flush.</summary>
private void CancelTimer()
{
_timerVersion++;
_timer?.Dispose();
_timer = null;
}
private void OnTimer(long version)
{
List<InboundPart> parts;
lock (_lock)
{
if (_disposed || version != _timerVersion || _slots.Any(s => s.Open)) return;
CancelTimer();
parts = _slots.Where(s => s.Part is not null).Select(s => s.Part!).ToList();
_slots.Clear();
if (parts.Count == 0) return;
_flushing++;
}
_ = DeliverAsync(parts);
}
private async Task DeliverAsync(List<InboundPart> parts)
{
try
{
await _onFlush(parts).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogError(ex, "Handing over {Count} message(s) failed", parts.Count);
}
finally
{
lock (_lock) _flushing--;
}
}
private int DropLocked()
{
_generation++;
CancelTimer();
var dropped = _slots.Count(s => s.Part is not null);
foreach (var slot in _slots) slot.Hold?.Dispose();
_slots.Clear();
return dropped;
}
}
/// <summary>One message's reserved place in a <see cref="ChatInbox"/> batch.</summary>
public sealed class InboundLease : IDisposable
{
private readonly ChatInbox _inbox;
private int _done;
internal InboundLease(ChatInbox inbox, ChatInbox.Slot slot, long generation, int messageId)
{
_inbox = inbox;
Slot = slot;
Generation = generation;
MessageId = messageId;
}
/// <summary>The Telegram message this lease was taken for.</summary>
public int MessageId { get; }
internal ChatInbox.Slot Slot { get; }
internal long Generation { get; }
/// <summary>Fill the slot; null means the message produced nothing. Only the first call counts.</summary>
public void Complete(InboundPart? part)
{
if (Interlocked.Exchange(ref _done, 1) == 0) _inbox.Complete(this, part);
}
/// <summary><c>Complete(null)</c> if not completed yet. Safe to call twice.</summary>
public void Dispose() => Complete(null);
}

206
tests/Volcanic.AgenticCore.Telegram.Tests/ChatInboxTests.cs
View File

@@ -1,3 +1,4 @@
using Microsoft.Extensions.Time.Testing;
using Volcanic.AgenticCore.Abstractions.Messages;
using Volcanic.AgenticCore.Telegram.Inbound;
@@ -13,6 +14,211 @@ public sealed class ChatInboxTests
new(id, At.AddSeconds(id - 1), group, reply, forward, body ?? [],
Enumerable.Range(0, images).Select(_ => new ImageContent("", "image/jpeg")).ToList());
private static readonly TimeSpan Window = TimeSpan.FromMilliseconds(600);
private static readonly TimeSpan Ms = TimeSpan.FromMilliseconds(1);
private readonly FakeTimeProvider _time = new(new DateTimeOffset(2026, 9, 26, 12, 0, 0, TimeSpan.Zero));
private readonly List<IReadOnlyList<InboundPart>> _flushes = [];
private ChatInbox Inbox(TimeSpan? maxHold = null) => new(parts =>
{
lock (_flushes) _flushes.Add(parts);
return Task.CompletedTask;
}, Window, _time, maxHold);
private static InboundPart Text(int id, string text = "x") => new(id, DateTimeOffset.UnixEpoch, null, null, null, [text], []);
/// <summary>Each flush as "1,2" (message ids in order).</summary>
private List<string> Batches { get { lock (_flushes) return _flushes.Select(f => string.Join(",", f.Select(p => p.MessageId))).ToList(); } }
[Fact]
public async Task One_part_flushes_after_the_window_not_before()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(Text(1));
_time.Advance(Window - Ms);
Assert.Empty(Batches);
_time.Advance(Ms);
Assert.Equal(["1"], Batches);
}
[Fact]
public async Task Two_parts_inside_the_window_make_one_flush_and_each_resets_the_timer()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(Text(1));
_time.Advance(TimeSpan.FromMilliseconds(400));
inbox.Begin(2).Complete(Text(2));
_time.Advance(TimeSpan.FromMilliseconds(400)); // 800 ms after the first, 400 after the second
Assert.Empty(Batches);
_time.Advance(TimeSpan.FromMilliseconds(200));
Assert.Equal(["1,2"], Batches);
}
[Fact]
public async Task Begin_cancels_an_armed_timer()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(Text(1));
_time.Advance(TimeSpan.FromMilliseconds(500));
var second = inbox.Begin(2);
_time.Advance(Window * 2);
Assert.Empty(Batches);
second.Complete(Text(2));
_time.Advance(Window);
Assert.Equal(["1,2"], Batches);
}
[Fact]
public async Task Complete_while_another_lease_is_open_does_not_arm()
{
await using var inbox = Inbox();
var first = inbox.Begin(1);
inbox.Begin(2).Complete(Text(2));
_time.Advance(Window * 3);
Assert.Empty(Batches);
first.Complete(Text(1));
_time.Advance(Window);
Assert.Equal(["1,2"], Batches);
}
[Fact]
public async Task Two_leases_need_two_completions()
{
await using var inbox = Inbox();
var a = inbox.Begin(1);
var b = inbox.Begin(2);
a.Complete(Text(1));
_time.Advance(Window * 3);
Assert.Empty(Batches);
Assert.Equal(1, inbox.OpenLeases);
b.Complete(Text(2));
_time.Advance(Window);
Assert.Equal(["1,2"], Batches);
}
[Fact]
public async Task A_null_part_never_flushes()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(null);
using (inbox.Begin(2)) { } // disposed without Complete: same as Complete(null)
_time.Advance(Window * 3);
Assert.Empty(Batches);
Assert.Equal(0, inbox.Buffered);
Assert.Equal(0, inbox.OpenLeases);
}
[Fact]
public async Task Comment_then_slow_voice_race_gives_one_flush()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(Text(1, "look at this")); // t=0.000 comment
_time.Advance(TimeSpan.FromMilliseconds(50));
var voice = inbox.Begin(2); // t=0.050 voice, transcription takes 3 s
_time.Advance(TimeSpan.FromSeconds(3));
Assert.Empty(Batches);
voice.Complete(Text(2, "[Voice message] hello")); // t=3.050
_time.Advance(Window); // t=3.650
Assert.Equal(["1,2"], Batches);
}
[Fact]
public async Task Voice_then_comment_race_keeps_slot_order()
{
await using var inbox = Inbox();
var voice = inbox.Begin(1);
_time.Advance(TimeSpan.FromMilliseconds(50));
inbox.Begin(2).Complete(Text(2, "what does it say?"));
_time.Advance(TimeSpan.FromSeconds(3));
voice.Complete(Text(1, "[Voice message] hello"));
_time.Advance(Window);
Assert.Equal(["1,2"], Batches); // Telegram order, not completion order
}
[Fact]
public async Task Discard_with_an_open_lease_drops_the_late_completion()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(Text(1));
var late = inbox.Begin(2);
inbox.Discard();
late.Complete(Text(2));
_time.Advance(Window * 2);
Assert.Empty(Batches);
Assert.Equal(0, inbox.Buffered);
inbox.Begin(3).Complete(Text(3));
_time.Advance(Window);
Assert.Equal(["3"], Batches);
}
[Fact]
public async Task Double_dispose_of_a_lease_is_a_no_op()
{
await using var inbox = Inbox();
var lease = inbox.Begin(1);
lease.Complete(Text(1));
lease.Dispose();
lease.Dispose();
var empty = inbox.Begin(2);
empty.Dispose();
empty.Dispose();
_time.Advance(Window);
Assert.Equal(["1"], Batches);
Assert.Equal(0, inbox.OpenLeases);
}
[Fact]
public async Task Max_hold_detaches_a_hung_lease_and_its_late_part_flushes_alone()
{
await using var inbox = Inbox(maxHold: TimeSpan.FromMinutes(2));
inbox.Begin(1).Complete(Text(1));
var hung = inbox.Begin(2);
_time.Advance(TimeSpan.FromMinutes(2)); // the hold expires: the batch no longer waits for #2
Assert.Empty(Batches);
_time.Advance(Window);
Assert.Equal(["1"], Batches);
hung.Complete(Text(2));
_time.Advance(Window);
Assert.Equal(["1", "2"], Batches);
}
[Fact]
public async Task Drain_completes_once_the_batch_is_handed_over()
{
await using var inbox = Inbox();
inbox.Begin(1).Complete(Text(1));
var drain = inbox.DrainAsync();
await Task.Delay(50);
Assert.False(drain.IsCompleted);
_time.Advance(Window);
await drain.WaitAsync(TimeSpan.FromSeconds(5));
}
[Fact]
public async Task Dispose_drops_buffered_parts()
{
var inbox = Inbox();
inbox.Begin(1).Complete(Text(1));
await inbox.DisposeAsync();
_time.Advance(Window);
Assert.Empty(Batches);
}
[Fact]
public void Render_starts_with_one_timestamp_header_from_the_first_part()
{

1
tests/Volcanic.AgenticCore.Telegram.Tests/Volcanic.AgenticCore.Telegram.Tests.csproj
View File

@@ -8,6 +8,7 @@
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
<PackageReference Include="xunit" Version="2.9.3" />
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.1" />
<PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.9.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="../../hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj" />