Skip to content

Commit 3baf3a2

Browse files
committed
fix(kernel): respect the outbox snapshot refresh interval
Avoid a full backlog scan after every successful claim. Prove continued dispatch with a retained snapshot before the configured interval and a refresh at its exact monotonic-clock boundary. ADR: 0046
1 parent 3620b42 commit 3baf3a2

3 files changed

Lines changed: 70 additions & 1 deletion

File tree

‎backend/src/LearnStack.Api/Messaging/OutboxProcessor.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
9999
_observation.Progress();
100100
}
101101

102-
await RefreshAsync(didWork, stoppingToken).ConfigureAwait(false);
102+
await RefreshAsync(force: false, stoppingToken).ConfigureAwait(false);
103103
}
104104
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
105105
{

‎backend/tests/LearnStack.Tests.Integration/Database/OutboxProcessorTests.cs‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,55 @@ public async Task Real_provisioning_rows_remain_Pending_across_hosted_polls_with
195195
(await test.ReadTerminalAuditsAsync()).Should().BeEmpty();
196196
}
197197

198+
[Fact]
199+
public async Task Successful_claims_keep_the_snapshot_until_its_interval_and_refresh_at_the_boundary()
200+
{
201+
await using var test = await WorkerDatabase.CreateAsync(schema.Postgres);
202+
var time = new SnapshotTimeProvider();
203+
var bounds = Bounds with
204+
{
205+
SnapshotInterval = TimeSpan.FromSeconds(1),
206+
PublishTimeout = TimeSpan.FromSeconds(10),
207+
LeaseDuration = TimeSpan.FromSeconds(30),
208+
ProgressTimeout = TimeSpan.FromSeconds(45),
209+
};
210+
test.Start(bounds, time);
211+
await UntilAsync(() => test.Observation.IsHealthy);
212+
var initial = test.Observation.Snapshot;
213+
initial.LastSnapshotTimestamp.Should().Be(0);
214+
215+
var first = await test.InsertAsync();
216+
await UntilAsync(async () => (await test.ReadAsync(first)).ProcessedAt is not null);
217+
var secondPartition = Guid.CreateVersion7();
218+
var second = test.Callbacks.Hold(secondPartition);
219+
var secondId = await test.InsertAsync(partition: secondPartition);
220+
await second.Entered.Task.WaitAsync(ProofDeadline);
221+
// Entering the next callback proves the preceding iteration (including
222+
// its refresh decision) completed. Wall-clock duration cannot expire this clock.
223+
test.Observation.Snapshot.Queue.Should().BeSameAs(initial.Queue);
224+
225+
time.Advance(bounds.SnapshotInterval - TimeSpan.FromTicks(1));
226+
second.ReleaseCallback.TrySetResult();
227+
await UntilAsync(async () => (await test.ReadAsync(secondId)).ProcessedAt is not null);
228+
var thirdPartition = Guid.CreateVersion7();
229+
var third = test.Callbacks.Hold(thirdPartition);
230+
var thirdId = await test.InsertAsync(partition: thirdPartition);
231+
await third.Entered.Task.WaitAsync(ProofDeadline);
232+
test.Observation.Snapshot.Queue.Should().BeSameAs(initial.Queue);
233+
test.Observation.Snapshot.LastSnapshotTimestamp.Should().Be(0);
234+
235+
time.Advance(TimeSpan.FromTicks(1));
236+
third.ReleaseCallback.TrySetResult();
237+
await UntilAsync(() => test.Observation.Snapshot.LastSnapshotTimestamp == time.GetTimestamp());
238+
239+
(await test.ReadAsync(thirdId)).ProcessedAt.Should().NotBeNull();
240+
test.Observation.Snapshot.Queue.Should().NotBeSameAs(initial.Queue);
241+
test.Observation.Snapshot.Queue.PendingCount.Should().Be(0);
242+
test.Observation.Snapshot.LastSnapshotTimestamp.Should().Be(bounds.SnapshotInterval.Ticks);
243+
test.Callbacks.Constructions.Should().Be(3);
244+
test.Observation.IsHealthy.Should().BeTrue();
245+
}
246+
198247
[Fact]
199248
public async Task A_poison_first_row_terminalizes_with_MUST_audit_and_the_following_real_handler_progresses()
200249
{
@@ -558,6 +607,14 @@ private sealed record Row(int Attempts, string? Token, DateTimeOffset? Processed
558607
DateTimeOffset AvailableAfter, string Type, string Payload);
559608
private sealed record TerminalAudit(string OperationType, string OperationClass, string Correlation, string Metadata);
560609

610+
private sealed class SnapshotTimeProvider : TimeProvider
611+
{
612+
private long _timestamp;
613+
public override long TimestampFrequency => TimeSpan.TicksPerSecond;
614+
public override long GetTimestamp() => Interlocked.Read(ref _timestamp);
615+
public void Advance(TimeSpan duration) => Interlocked.Add(ref _timestamp, duration.Ticks);
616+
}
617+
561618
/// <summary>Controls shutdown deadlines while all other host timers keep running normally.</summary>
562619
private sealed class ShutdownTimeProvider(TimeSpan shutdownTimeout) : TimeProvider
563620
{

‎docs/roadmap/phase-02b-events-auth.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2050,6 +2050,18 @@ Both correction rounds and local validation are complete. Exact correction-head
20502050
CI remains pending in this local receipt; the PR description owns live checks
20512051
and artifact-backed evidence after publication.
20522052

2053+
#### Snapshot interval correction (2026-10-13)
2054+
2055+
The dispatch loop no longer forces a full queue snapshot after every successful
2056+
claim. It checks `SnapshotInterval` after completed iterations, while `didWork`
2057+
still controls idle polling. A controlled monotonic-clock integration case
2058+
dispatches three real rows: the initial snapshot survives successful work and
2059+
the last tick before the interval; the exact boundary refreshes it after the
2060+
next completed dispatch. Release build reports zero warnings/errors; format,
2061+
1,847 unit, 287 architecture and all 17 worker integration cases pass, with zero
2062+
failures/skips. Independent correction reviews and new-head CI remain pending
2063+
in this initial local receipt.
2064+
20532065
### Durable outbox dispatch
20542066

20552067
P02b-1 Steps 2/3 implement `IOutbox`, replace Packet 3's step-7 shell with

0 commit comments

Comments
 (0)