From 1bf1fd32f993e6acfd4d934deaa18395adbe0ea4 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 18 Aug 2026 15:49:18 +0000 Subject: [PATCH] Defer busy current-session reminders --- .../.system/files/netclaw-operations/SKILL.md | 2 +- .../references/scheduling.md | 9 + .../.openspec.yaml | 2 + .../design.md | 78 ++++++++ .../proposal.md | 36 ++++ .../specs/netclaw-scheduling/spec.md | 81 +++++++++ .../tasks.md | 21 +++ .../Contracts/SessionBindingContractTests.cs | 61 +++++-- .../TestHelpers/RecordingSessionPipeline.cs | 12 ++ .../Reminders/ReminderExecutionActorTests.cs | 58 ++++++ .../Reminders/ReminderManagerActorTests.cs | 135 ++++++++++++++ .../Reminders/ReminderExecutionActor.cs | 40 ++++- .../Reminders/ReminderManagerActor.cs | 129 ++++++++++++++ .../Reminders/ReminderProtocol.cs | 8 + .../Sessions/SessionProtocol.Responses.cs | 9 + .../Sessions/SessionProtocol.cs | 3 +- .../DiscordSessionBindingActor.cs | 43 +++-- .../MattermostSessionBindingActor.cs | 43 +++-- .../SlackThreadBindingActor.cs | 43 +++-- .../Gateway/SignalRSessionActorTests.cs | 166 ++++++++++++++++++ .../Gateway/SignalRSessionActor.cs | 47 +++-- 21 files changed, 942 insertions(+), 84 deletions(-) create mode 100644 openspec/changes/defer-busy-current-session-reminders/.openspec.yaml create mode 100644 openspec/changes/defer-busy-current-session-reminders/design.md create mode 100644 openspec/changes/defer-busy-current-session-reminders/proposal.md create mode 100644 openspec/changes/defer-busy-current-session-reminders/specs/netclaw-scheduling/spec.md create mode 100644 openspec/changes/defer-busy-current-session-reminders/tasks.md create mode 100644 src/Netclaw.Daemon.Tests/Gateway/SignalRSessionActorTests.cs diff --git a/feeds/skills/.system/files/netclaw-operations/SKILL.md b/feeds/skills/.system/files/netclaw-operations/SKILL.md index bf6168709..02016c477 100644 --- a/feeds/skills/.system/files/netclaw-operations/SKILL.md +++ b/feeds/skills/.system/files/netclaw-operations/SKILL.md @@ -3,7 +3,7 @@ name: netclaw-operations description: "REQUIRED when the user asks about scheduling, reminders, cron jobs, timers, background jobs, diagnostics, troubleshooting, MCP tools, daemon health, identity updates, or Netclaw capabilities and self-maintenance." metadata: author: netclaw - version: "2.58.0" + version: "2.59.0" --- # Netclaw Operations diff --git a/feeds/skills/.system/files/netclaw-operations/references/scheduling.md b/feeds/skills/.system/files/netclaw-operations/references/scheduling.md index e63f0a480..653942f1c 100644 --- a/feeds/skills/.system/files/netclaw-operations/references/scheduling.md +++ b/feeds/skills/.system/files/netclaw-operations/references/scheduling.md @@ -92,6 +92,15 @@ A known execution or delivery failure starts the Akka.Reminders retry policy. The retry uses bounded backoff and the same durable occurrence identity. A successful execution resets the consecutive failure count. +A `current_session` reminder requires a session with no active turn. The session +defers the reminder during an active turn, a restart drain, or pipeline outage. +A supported gateway also defers the reminder when startup has not registered it. +Netclaw sends an Akka.Reminders negative acknowledgement for each deferral. +Each negative acknowledgement consumes one delivery attempt and applies durable +backoff. A scheduled retry does not increase the Netclaw failure count or create +failure history. Retry exhaustion creates one terminal failure and disables the +reminder. An unsupported origin channel is a permanent delivery failure. + A one-shot reminder stays enabled while an occurrence can retry. After a successful acknowledgement, Netclaw deletes its definition and history. A poison one-shot becomes disabled with a `Failed` outcome. Its definition and history diff --git a/openspec/changes/defer-busy-current-session-reminders/.openspec.yaml b/openspec/changes/defer-busy-current-session-reminders/.openspec.yaml new file mode 100644 index 000000000..95672402a --- /dev/null +++ b/openspec/changes/defer-busy-current-session-reminders/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-18 diff --git a/openspec/changes/defer-busy-current-session-reminders/design.md b/openspec/changes/defer-busy-current-session-reminders/design.md new file mode 100644 index 000000000..b13d8d35b --- /dev/null +++ b/openspec/changes/defer-busy-current-session-reminders/design.md @@ -0,0 +1,78 @@ +## Context + +CurrentSession reminders reuse the origin session and its channel binding. Each binding writes trusted reminder input into the same session queue as user input. + +The session accepts queued input during an active turn. Its tool-loop drain then copies only message content into the active turn and discards reminder identity. + +The channel binding registers a delivery observer before queue admission. The observer waits for a `TurnCompleted.SourceReminderId` that the active human turn cannot produce. + +Akka.Reminders 0.7.0 has no separate defer API. Its `NackAsync` contract schedules durable retry with bounded exponential backoff and consumes one delivery attempt. + +## Goals / Non-Goals + +**Goals:** + +- Reject CurrentSession reminder admission while the target session has an active turn. +- Use Akka.Reminders for durable delay and retry ownership. +- Keep transient deferrals out of Netclaw failure history, alerts, and poison counts. +- Convert retry-budget exhaustion into one real reminder failure. +- Apply one contract to every supported CurrentSession channel binding. + +**Non-Goals:** + +- Change ordinary user-message buffer behavior. +- Change Channel or None reminder execution. +- Add a new Akka.Reminders API. +- Change reminder health-count retention or decay. +- Change reminder trust or approval policy. + +## Decisions + +### Channel bindings own the admission check + +Slack, Discord, and Mattermost bindings already track `_turnInFlight`. They will reject a trusted reminder before observer registration or queue admission. + +SignalR will track the same state because it has no equivalent field. Each binding will set the state after queue admission and clear it after turn completion or pipeline reset. + +This boundary prevents metadata loss and observer leaks. A session-actor check would occur after the binding registers its delivery observer. + +### A typed response distinguishes deferral from rejection + +`CommandDeferred(SessionId, Reason)` will extend `ISessionResponse`. `CommandNack` will continue to identify permanent rejection. + +The execution actor will map `CommandDeferred` to an internal `ReminderExecutionDeferred`. It will not create a failed history record. + +### The manager maps deferral to durable negative acknowledgement + +The manager will call `NackAsync` with the original envelope and the deferral reason. `RetryScheduled` will release the active execution without Netclaw failure state. + +`Failed` or `Expired` means the scheduler cannot retry. The manager will then record one failed history entry and apply the existing terminal failure policy. + +This choice consumes Akka delivery attempts. The existing policy provides ten attempts with backoff from one minute to a ten-minute cap. + +### Supported gateway absence is transient + +The execution actor will distinguish an unsupported origin from a supported gateway that has not registered. The supported case will use the deferral path. + +The next Akka attempt will resolve the registry again. Netclaw will not add a local poll loop or retain a second retry queue. + +### Accepted turns keep the delivery contract + +After admission, `CommandAck`, `ReminderDeliveryResult`, and the one-hour observation timeout keep their current meaning. Transport delivery failures remain execution failures. + +## Risks / Trade-offs + +- **A session can stay busy through all ten attempts.** The final result becomes one terminal failure. +- **A queue-write timeout can race with successful admission.** Stable reminder identity lets the later Akka attempt use session deduplication. +- **A binding-local busy flag can become stale after a stream fault.** Every pipeline reset clears the flag and fails registered observers. +- **The terminal deferral result arrives after Akka settles the occurrence.** The manager will report any later local persistence failure as a settlement fault. + +## Migration Plan + +1. Deploy the additive transient response and manager behavior together. +2. Keep the existing reminder JSON and Akka.Reminders database formats. +3. Roll back by restoring the prior binary. No data migration is necessary. + +## Open Questions + +None. diff --git a/openspec/changes/defer-busy-current-session-reminders/proposal.md b/openspec/changes/defer-busy-current-session-reminders/proposal.md new file mode 100644 index 000000000..00b3a73ad --- /dev/null +++ b/openspec/changes/defer-busy-current-session-reminders/proposal.md @@ -0,0 +1,36 @@ +## Why + +PRD-008 requires reliable reminder retries and useful failure signals. A busy session now accepts a CurrentSession reminder but cannot preserve its turn identity. + +The reminder then waits one hour for an observation that cannot occur. A late gateway registration also creates a false permanent failure. + +## What Changes + +- Add a typed transient deferral response for CurrentSession reminder admission. +- Make each supported session binding defer a reminder before queue admission when its session has an active turn. +- Make the reminder execution actor report transient admission deferrals to the reminder manager. +- Make the manager use the Akka.Reminders negative acknowledgement path for durable backoff. +- Do not write failed history, increment `ConsecutiveFailures`, or emit failure alerts while Akka schedules another attempt. +- Count retry-budget exhaustion as one real reminder failure. +- Treat late registration for a supported origin gateway as a deferral. +- Keep permanent validation errors and accepted-turn delivery failures on the existing failure path. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `netclaw-scheduling`: Define transient CurrentSession admission deferral and its retry and failure-count behavior. + +## Impact + +- **Source PRDs:** PRD-008. +- **In scope:** CurrentSession delivery through Slack, Discord, Mattermost, SignalR, and TUI session bindings. +- **Out of scope:** Channel delivery, no-delivery reminders, reminder health-count decay, and a new Akka.Reminders deferral API. +- **APIs:** Add one transient session response. No wire or persistence format changes apply. +- **Dependencies:** Continue to use `Aaron.Akka.Reminders` 0.7.0 and its existing `NackAsync` retry contract. +- **Security:** Preserve the current trusted reminder source, audience, boundary, and approval policy. +- **Operations:** Busy sessions cause bounded scheduler backoff. Scheduled deferrals do not appear as Netclaw execution failures. diff --git a/openspec/changes/defer-busy-current-session-reminders/specs/netclaw-scheduling/spec.md b/openspec/changes/defer-busy-current-session-reminders/specs/netclaw-scheduling/spec.md new file mode 100644 index 000000000..343e6ddac --- /dev/null +++ b/openspec/changes/defer-busy-current-session-reminders/specs/netclaw-scheduling/spec.md @@ -0,0 +1,81 @@ +## ADDED Requirements + +### Requirement: Transient CurrentSession admission deferral + +The CurrentSession delivery path SHALL defer a trusted reminder before queue admission when the target session cannot start a distinct turn. Deferral SHALL use the original Akka.Reminders occurrence and SHALL consume one Akka delivery attempt. + +#### Scenario: Busy session defers before queue admission + +- **GIVEN** a supported CurrentSession binding has an active turn +- **WHEN** the binding receives `DeliverTrustedSessionTurn` +- **THEN** it replies with `CommandDeferred` +- **AND** it does not register a delivery observer +- **AND** it does not write the reminder to the session queue + +#### Scenario: Successful admission marks the binding busy + +- **GIVEN** a supported CurrentSession binding has no active turn +- **WHEN** it admits a trusted reminder to the session queue +- **THEN** it marks the turn as active before it handles another admission +- **AND** it clears the active state after `TurnCompleted` or pipeline reset + +#### Scenario: Supported gateway has not registered + +- **GIVEN** a CurrentSession reminder has a supported origin channel type +- **AND** that channel gateway has not registered after daemon startup +- **WHEN** the execution actor resolves the gateway +- **THEN** it reports a transient deferral +- **AND** it does not report an unsupported channel error + +#### Scenario: Unsupported origin remains a failure + +- **GIVEN** a CurrentSession reminder has an unsupported origin channel type +- **WHEN** the execution actor validates the origin +- **THEN** it reports a permanent execution failure + +### Requirement: Deferred occurrence settlement + +The reminder manager SHALL settle a transient admission deferral through `IReminderClient.NackAsync`. It SHALL separate an available scheduler retry from a terminal retry result. + +#### Scenario: Scheduler accepts the deferral + +- **GIVEN** a CurrentSession execution reports a transient deferral +- **WHEN** `NackAsync` returns `RetryScheduled` +- **THEN** the manager releases the active execution +- **AND** it does not append a failed history record +- **AND** it does not increment `ConsecutiveFailures` +- **AND** it does not emit a reminder failure alert +- **AND** reminder status exposes the scheduler's next attempt + +#### Scenario: Deferral exhausts the retry budget + +- **GIVEN** a CurrentSession execution reports a transient deferral +- **WHEN** `NackAsync` returns `Failed` or `Expired` +- **THEN** the manager records one failed history entry +- **AND** it increments `ConsecutiveFailures` once +- **AND** it applies the existing terminal occurrence policy + +#### Scenario: Retry enters an idle session + +- **GIVEN** Akka.Reminders retries a deferred CurrentSession occurrence +- **AND** the target session is idle +- **WHEN** the binding admits the reminder +- **THEN** the reminder runs as a distinct turn +- **AND** `TurnCompleted.SourceReminderId` contains the stable occurrence key + +#### Scenario: Accepted-turn delivery fails + +- **GIVEN** a CurrentSession reminder was admitted as a distinct turn +- **WHEN** its channel reports `ReminderDeliveryResult.Delivered` as false +- **THEN** the manager records the result through the existing failure path + +### Requirement: CurrentSession deferral channel parity + +Slack, Discord, Mattermost, SignalR, and TUI session delivery SHALL apply the same transient deferral contract. + +#### Scenario: Each supported binding rejects concurrent reminder admission + +- **GIVEN** any supported CurrentSession channel binding has an active turn +- **WHEN** another CurrentSession reminder targets that binding +- **THEN** the binding replies with `CommandDeferred` +- **AND** the scheduler owns the next attempt diff --git a/openspec/changes/defer-busy-current-session-reminders/tasks.md b/openspec/changes/defer-busy-current-session-reminders/tasks.md new file mode 100644 index 000000000..8875cf214 --- /dev/null +++ b/openspec/changes/defer-busy-current-session-reminders/tasks.md @@ -0,0 +1,21 @@ +## 1. Admission contract + +- [x] 1.1 Add the transient session response and reminder execution outcome. +- [x] 1.2 Defer busy or unavailable CurrentSession admission across all supported bindings. +- [x] 1.3 Distinguish a late supported gateway from an unsupported origin. + +## 2. Durable settlement + +- [x] 2.1 Map transient deferral to Akka.Reminders `NackAsync` without Netclaw failure state. +- [x] 2.2 Convert retry-budget exhaustion into one terminal reminder failure. + +## 3. Automated proof + +- [x] 3.1 Add binding contract tests for busy deferral and observer cleanup. +- [x] 3.2 Add reminder manager tests for scheduled deferral and terminal exhaustion. +- [x] 3.3 Add execution actor tests for late gateway registration and unsupported origins. + +## 4. Guidance and validation + +- [x] 4.1 Update the scheduling operations guidance and increase its skill version. +- [x] 4.2 Run focused tests, affected suites, Slopwatch, header verification, and `git diff --check`. diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs index 0a4356eda..06520d10d 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs @@ -323,43 +323,68 @@ public async Task Reminder_delivery_reports_failure_when_post_throws() ClearReplyClientThrows(); } - // Regression for the observer-clobber bug: two distinct reminders can target - // the same session concurrently. A single observer field is overwritten by - // the second dispatch before the first turn completes, so the first - // reminder's result is misrouted. Each observer must receive ITS OWN keyed - // result. + // A busy session must defer a second reminder before it registers an observer. + // The same occurrence can enter the session after the active turn ends. [Fact] - public async Task Concurrent_reminders_to_same_session_each_get_their_own_result() + public async Task Busy_session_defers_a_second_reminder() { var ct = TestContext.Current.CancellationToken; var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); var sid = new SessionId("session-reminder-concurrent"); const string keyA = "reminder-A:111"; const string keyB = "reminder-B:222"; + var outputRelease = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); var pipeline = new RecordingSessionPipeline(_ => [ - new TextOutput("reply A") { SessionId = sid }, - new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = new ReminderId(keyA) }, - new TextOutput("reply B") { SessionId = sid }, - new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(2), SourceReminderId = new ReminderId(keyB) } - ], reactive: true); + new TurnCompleted + { + SessionId = sid, + TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), + SourceReminderId = new ReminderId(keyA) + }, + new TurnCompleted + { + SessionId = sid, + TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(2), + SourceReminderId = new ReminderId(keyB) + }, + new TextOutput("retry barrier") { SessionId = sid } + ], outputRelease.Task); var observerA = CreateTestProbe(); var observerB = CreateTestProbe(); + var secondAttempt = CreateTestProbe(); var binding = CreateBindingActor(sid, pipeline, detector); await pipeline.Created; - // Both reminders dispatched before either turn completes. - binding.Tell(new DeliverTrustedSessionTurn(sid, "reminder A", CreateReminderSource(keyA, observerA.Ref))); - binding.Tell(new DeliverTrustedSessionTurn(sid, "reminder B", CreateReminderSource(keyB, observerB.Ref))); + binding.Tell(new DeliverTrustedSessionTurn( + sid, + "reminder A", + CreateReminderSource(keyA, observerA.Ref))); + await AwaitAssertAsync(() => Assert.Single(pipeline.CapturedInputs), cancellationToken: ct); - var resultA = await observerA.ExpectMsgAsync( + secondAttempt.Send( + binding, + new DeliverTrustedSessionTurn(sid, "reminder B", CreateReminderSource(keyB, observerB.Ref))); + var deferred = await secondAttempt.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: ct); - Assert.Equal(new ReminderId(keyA), resultA.ReminderDeliveryKey); + Assert.Equal(sid, deferred.SessionId); + Assert.Contains("active turn", deferred.Reason, StringComparison.OrdinalIgnoreCase); - var resultB = await observerB.ExpectMsgAsync( + outputRelease.TrySetResult(); + var firstResult = await observerA.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: ct); - Assert.Equal(new ReminderId(keyB), resultB.ReminderDeliveryKey); + Assert.Equal(new ReminderId(keyA), firstResult.ReminderDeliveryKey); + await observerB.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(250), ct); + await AwaitAssertAsync(() => + { + Assert.Contains(GetPostedTexts(), text => + text.Contains("retry barrier", StringComparison.Ordinal)); + }, cancellationToken: ct); + + var input = Assert.Single(pipeline.CapturedInputs); + Assert.Equal(new ReminderId(keyA), input.ReminderId); } // Regression for the misleading-fallback bug: when the real content post diff --git a/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSessionPipeline.cs b/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSessionPipeline.cs index 685c967c6..0098c6730 100644 --- a/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSessionPipeline.cs +++ b/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSessionPipeline.cs @@ -21,6 +21,7 @@ public sealed class RecordingSessionPipeline : ISessionPipeline private readonly List _recordedFeedback = []; private readonly Func> _outputFactory; private readonly bool _reactive; + private readonly Task? _outputRelease; private readonly TaskCompletionSource _created = new( TaskCreationOptions.RunContinuationsAsynchronously); private SessionPipelineOptions? _capturedOptions; @@ -49,6 +50,15 @@ public RecordingSessionPipeline( _reactive = reactive; } + public RecordingSessionPipeline( + Func> outputFactory, + Task outputRelease) + { + _outputFactory = outputFactory; + _reactive = true; + _outputRelease = outputRelease; + } + public SessionPipelineOptions? CapturedOptions => Volatile.Read(ref _capturedOptions); public Task Created => _created.Task; public IReadOnlyList RecordedFeedback @@ -105,6 +115,8 @@ public Task CreateAsync( { // Wait for the first input to arrive before emitting anything. await gate.Reader.ReadAsync(cancellationToken); + if (_outputRelease is not null) + await _outputRelease.WaitAsync(cancellationToken); } if (state < outputs.Count) diff --git a/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs b/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs index a4ccccbab..c4985b531 100644 --- a/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs +++ b/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs @@ -373,6 +373,44 @@ public async Task Execution_fails_at_absolute_attempt_limit_despite_output_activ } } + [Fact] + public async Task Missing_supported_gateway_defers_CurrentSession_delivery() + { + var definition = CreateCurrentSessionDefinition("missing-slack", ChannelType.Slack); + var pipeline = new FailingSessionPipeline(new InvalidOperationException("pipeline must not start")); + var probe = CreateTestProbe(); + + Sys.ActorOf( + Props.Create(() => new ParentProxy(probe.Ref, definition, pipeline, _historyStore)), + "exec-missing-supported-gateway"); + + var deferred = await probe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + + Assert.Equal(definition.Id, deferred.Id); + Assert.Contains("not registered", deferred.Reason, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public async Task Unsupported_gateway_fails_CurrentSession_delivery() + { + var definition = CreateCurrentSessionDefinition("unsupported-origin", ChannelType.Reminder); + var pipeline = new FailingSessionPipeline(new InvalidOperationException("pipeline must not start")); + var probe = CreateTestProbe(); + + Sys.ActorOf( + Props.Create(() => new ParentProxy(probe.Ref, definition, pipeline, _historyStore)), + "exec-unsupported-gateway"); + + var completed = await probe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + + Assert.False(completed.Success); + Assert.Contains("does not support", completed.ErrorMessage!, StringComparison.OrdinalIgnoreCase); + } + private static ReminderDefinition CreateDefinition(string id) { var now = TimeProvider.System.GetUtcNow(); @@ -397,6 +435,20 @@ private static ReminderDefinition CreateDefinition(string id) }; } + private static ReminderDefinition CreateCurrentSessionDefinition(string id, ChannelType originChannelType) + { + var definition = CreateDefinition(id); + return definition with + { + Delivery = new ReminderDelivery + { + Kind = DeliveryKind.CurrentSession, + SessionId = "C0123ABC/1712000000.000001", + OriginChannelType = originChannelType + } + }; + } + /// /// Minimal parent actor that creates as a child /// and forwards messages it receives to a probe for test assertions. @@ -436,6 +488,12 @@ public ParentProxy( if (acceptCompletion) Sender.Tell(new ReminderExecutionAccepted(completed.ExecutionId)); }); + Receive(deferred => + { + probe.Tell(deferred); + if (acceptCompletion) + Sender.Tell(new ReminderExecutionAccepted(deferred.ExecutionId)); + }); ReceiveAny(msg => probe.Tell(msg)); } } diff --git a/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs b/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs index 9cd3f6a5a..74614f5f6 100644 --- a/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs +++ b/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs @@ -930,6 +930,119 @@ await AwaitAssertAsync(async () => } } + [Fact] + public async Task CurrentSession_deferral_retries_without_a_Netclaw_failure() + { + var manager = await GetManagerAsync(); + var gatewayProbe = CreateTestProbe("defer-once-gateway"); + var gateway = Sys.ActorOf( + Props.Create(() => new DeferringTrustedGateway(gatewayProbe.Ref, deferredAttempts: 1)), + "defer-once-current-session-gateway"); + ActorRegistry.For(Sys).Register(gateway); + + var now = _timeProvider.GetUtcNow(); + var definition = CreateCurrentSessionDefinition("defer-once", deliveryRequired: false) with + { + Schedule = new ReminderSchedule + { + Type = ReminderScheduleType.Interval, + FireAt = now.AddMilliseconds(100), + IntervalTicks = TimeSpan.FromHours(2).Ticks + } + }; + + var saved = await manager.Ask( + new SaveReminderCommand( + definition, + Authorization: new ReminderAudienceAuthorizationContext(TrustAudience.Team, "test")), + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + Assert.True(saved.Success, saved.ErrorMessage); + + await gatewayProbe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await gatewayProbe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + await AwaitAssertAsync(async () => + { + var status = await manager.Ask( + new GetReminderStatusQuery(definition.Id), + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + Assert.True(status.Found); + Assert.True(status.Enabled); + Assert.Equal(0, status.ConsecutiveFailures); + Assert.Contains(status.RecentHistory, history => history.Success); + + var health = await manager.Ask( + GetReminderHealthQuery.Instance, + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + Assert.Equal(0, health.FailedCount); + }, duration: TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + Assert.DoesNotContain(_notificationSink.Alerts, alert => + alert.Category == AlertType.ReminderExecutionFailed + && alert.Source == definition.Id.Value); + } + + [Fact] + public async Task CurrentSession_deferral_records_one_failure_after_retry_exhaustion() + { + var manager = await GetManagerAsync(); + var gatewayProbe = CreateTestProbe("always-defer-gateway"); + var gateway = Sys.ActorOf( + Props.Create(() => new DeferringTrustedGateway(gatewayProbe.Ref, deferredAttempts: int.MaxValue)), + "always-defer-current-session-gateway"); + ActorRegistry.For(Sys).Register(gateway); + + var definition = CreateCurrentSessionDefinition("defer-terminal", deliveryRequired: false) with + { + Schedule = new ReminderSchedule + { + Type = ReminderScheduleType.OneShot, + FireAt = _timeProvider.GetUtcNow().AddMilliseconds(100) + } + }; + + var saved = await manager.Ask( + new SaveReminderCommand( + definition, + Authorization: new ReminderAudienceAuthorizationContext(TrustAudience.Team, "test")), + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + Assert.True(saved.Success, saved.ErrorMessage); + + await gatewayProbe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + await AwaitAssertAsync(async () => + { + var status = await manager.Ask( + new GetReminderStatusQuery(definition.Id), + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + Assert.True(status.Found); + Assert.False(status.Enabled); + Assert.Equal(1, status.ConsecutiveFailures); + Assert.Equal(ReminderTerminalOutcome.Failed, status.TerminalOutcome); + var failure = Assert.Single(status.RecentHistory); + Assert.False(failure.Success); + Assert.Contains("active turn", failure.ErrorMessage, StringComparison.OrdinalIgnoreCase); + + var health = await manager.Ask( + GetReminderHealthQuery.Instance, + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + Assert.Equal(1, health.FailedCount); + }, duration: TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + Assert.Single(_notificationSink.Alerts, alert => + alert.Category == AlertType.ReminderExecutionFailed + && alert.Source == definition.Id.Value); + } + [Fact] public async Task Reconcile_disables_expired_recurring_reminders() { @@ -1454,6 +1567,28 @@ public AutoAckTrustedGateway(IActorRef probe) } } + private sealed class DeferringTrustedGateway : ReceiveActor + { + private readonly IActorRef _probe; + private readonly int _deferredAttempts; + private int _attemptCount; + + public DeferringTrustedGateway(IActorRef probe, int deferredAttempts) + { + _probe = probe; + _deferredAttempts = deferredAttempts; + + Receive(msg => + { + _probe.Tell(msg); + _attemptCount++; + Sender.Tell(_attemptCount <= _deferredAttempts + ? CommandDeferred.For(msg.SessionId, "The target session has an active turn.") + : CommandAck.For(msg.SessionId)); + }); + } + } + private sealed class FailingReminderSessionPipeline(string reason) : ISessionPipeline { private int _invocationCount; diff --git a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs index 38897dbcf..1ec19f8ea 100644 --- a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs +++ b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs @@ -272,7 +272,14 @@ private async Task InitializeCurrentSessionAsync() var gateway = ResolveGatewayFor(originChannelType); if (gateway is null) { - ReportOutcome(false, $"Mode B unsupported origin channel type: {originChannelType}"); + if (IsSupportedCurrentSessionOrigin(originChannelType)) + { + ReportDeferred($"The {originChannelType} gateway is not registered yet."); + } + else + { + ReportOutcome(false, $"CurrentSession does not support origin channel type: {originChannelType}"); + } return; } @@ -310,6 +317,13 @@ private async Task InitializeCurrentSessionAsync() ReportOutcome(false, $"Session rejected reminder delivery: {nack.Reason}"); break; + case CommandDeferred deferred: + _log.Info( + "reminder_current_session_deferred execution_id={ExecutionId} reminder_id={ReminderId} session_id={SessionId} reason={Reason}", + _executionId, _definition.Id, sessionId.Value, deferred.Reason); + ReportDeferred(deferred.Reason); + break; + default: _log.Warning( "reminder_current_session_unexpected_reply execution_id={ExecutionId} reminder_id={ReminderId} reply_type={ReplyType}", @@ -412,6 +426,13 @@ private void HandleDeliveryBackstopTimeout(DeliveryBackstopTimeout msg) }; } + private static bool IsSupportedCurrentSessionOrigin(ChannelType originChannelType) => + originChannelType is ChannelType.Slack + or ChannelType.Discord + or ChannelType.Mattermost + or ChannelType.Tui + or ChannelType.SignalR; + private TrustBoundary GetPersistedBoundaryOrThrow() { if (!SecurityPolicyDefaults.TryNormalizeBoundary(_definition.Boundary.Value, out var normalizedBoundary)) @@ -602,6 +623,23 @@ private void ReportOutcome(bool success, string? errorMessage = null) errorMessage)); } + private void ReportDeferred(string reason) + { + if (_completed || _settlementStarted) + return; + + _settlementStarted = true; + Context.SetReceiveTimeout(null); + Timers.Cancel(ExecutionAttemptTimerKey); + Timers.Cancel(DeliveryBackstopTimerKey); + _completed = true; + + Context.Parent.Tell(new ReminderExecutionDeferred( + _executionId, + _definition.Id, + reason)); + } + private void HandleExecutionAccepted(ReminderExecutionAccepted accepted) { if (!_completed || accepted.ExecutionId != _executionId) diff --git a/src/Netclaw.Actors/Reminders/ReminderManagerActor.cs b/src/Netclaw.Actors/Reminders/ReminderManagerActor.cs index 9d5a27edb..f46252cf3 100644 --- a/src/Netclaw.Actors/Reminders/ReminderManagerActor.cs +++ b/src/Netclaw.Actors/Reminders/ReminderManagerActor.cs @@ -82,6 +82,7 @@ public ReminderManagerActor( ReceiveAsync>(HandleReminderFiredAsync); ReceiveAsync(HandleExecutionOutcomeAsync); + ReceiveAsync(HandleExecutionDeferredAsync); ReceiveAsync(HandleExecutionTerminatedAsync); ReceiveAsync(_ => HandleReconcileAsync()); @@ -755,6 +756,134 @@ private async Task HandleExecutionOutcomeAsync(ReminderExecutionCompleted outcom } } + private async Task HandleExecutionDeferredAsync(ReminderExecutionDeferred deferred) + { + var replyTo = Sender; + if (!_activeExecutions.TryGet(deferred.Id, out var execution) + || execution.ExecutionId != deferred.ExecutionId) + { + replyTo.Tell(new ReminderExecutionAccepted(deferred.ExecutionId)); + return; + } + + try + { + await SettleDeferredExecutionAsync(deferred, execution); + } + catch (Exception ex) + { + _log.Error(ex, "Unexpected reminder deferral failure for '{0}'", deferred.Id.Value); + var definition = _definitionStore.Get(deferred.Id); + if (definition is not null) + EmitSettlementFailure(definition, ex.Message, ex); + } + finally + { + _activeExecutions.TryRemove(deferred.Id, deferred.ExecutionId, out _); + replyTo.Tell(new ReminderExecutionAccepted(deferred.ExecutionId)); + } + } + + private async Task SettleDeferredExecutionAsync( + ReminderExecutionDeferred deferred, + ActiveReminderExecution execution) + { + var definition = _definitionStore.Get(deferred.Id); + AkkaReminderProtocol.ReminderNackResponse nack; + try + { + nack = await _client!.NackAsync(execution.Envelope, deferred.Reason); + } + catch (Exception ex) + { + if (definition is not null) + EmitSettlementFailure(definition, ex.Message, ex); + return; + } + + switch (nack.ResponseCode) + { + case ReminderNackResponseCode.RetryScheduled: + _log.Info( + "Reminder '{0}' deferred because its session is unavailable. The scheduler will retry at {1}.", + deferred.Id.Value, + nack.NextAttemptAtUtc); + return; + + case ReminderNackResponseCode.Failed: + case ReminderNackResponseCode.Expired: + await RecordTerminalDeferralAsync(deferred, execution, definition); + return; + + case ReminderNackResponseCode.NotFound: + _log.Info( + "Reminder '{0}' deferral found no active occurrence. The scheduler already settled it.", + deferred.Id.Value); + return; + + case ReminderNackResponseCode.Error: + if (definition is not null) + { + EmitSettlementFailure( + definition, + nack.Message ?? $"Negative acknowledgement returned {nack.ResponseCode}."); + } + return; + + default: + throw new ArgumentOutOfRangeException( + nameof(nack.ResponseCode), + nack.ResponseCode, + "Unexpected reminder deferral response."); + } + } + + private async Task RecordTerminalDeferralAsync( + ReminderExecutionDeferred deferred, + ActiveReminderExecution execution, + ReminderDefinition? definition) + { + var now = _timeProvider.GetUtcNow(); + var sessionId = definition?.Delivery.SessionId + ?? $"reminder/{deferred.Id}/{execution.Envelope.DueTimeUtc.ToUnixTimeMilliseconds()}"; + var history = new HistoryRecord( + execution.StartedAt, + Success: false, + DurationMs: (long)(now - execution.StartedAt).TotalMilliseconds, + sessionId, + deferred.Reason); + await AppendHistorySafelyAsync(deferred.Id, history); + + var count = (definition?.ConsecutiveFailures ?? 0) + 1; + if (definition is not null) + { + try + { + definition = definition with + { + ConsecutiveFailures = count, + Enabled = false, + TerminalOutcome = ReminderTerminalOutcome.Failed, + UpdatedAtMs = now.ToUnixTimeMilliseconds() + }; + _definitionStore.Save(definition); + } + catch (Exception ex) + { + EmitSettlementFailure(definition, ex.Message, ex); + return; + } + } + + ReportExecutionFailure( + deferred.Id, + definition, + count, + deferred.Reason, + willDisable: true); + await CancelScheduleOnlyAsync(deferred.Id); + } + private async Task HandleExecutionTerminatedAsync(ReminderExecutionTerminated terminated) { if (!_activeExecutions.TryRemove(terminated.Id, terminated.ExecutionId, out var execution)) diff --git a/src/Netclaw.Actors/Reminders/ReminderProtocol.cs b/src/Netclaw.Actors/Reminders/ReminderProtocol.cs index b43716ba9..1fb0cf6bf 100644 --- a/src/Netclaw.Actors/Reminders/ReminderProtocol.cs +++ b/src/Netclaw.Actors/Reminders/ReminderProtocol.cs @@ -477,6 +477,14 @@ internal sealed record ReminderExecutionCompleted( HistoryRecord History, string? ErrorMessage = null) : INoSerializationVerificationNeeded; +/// +/// Sent when a CurrentSession target cannot accept a distinct turn now. +/// +internal sealed record ReminderExecutionDeferred( + Guid ExecutionId, + ReminderId Id, + string Reason) : INoSerializationVerificationNeeded; + internal sealed record ReminderExecutionAccepted(Guid ExecutionId) : INoSerializationVerificationNeeded; internal sealed record ReminderExecutionTerminated( diff --git a/src/Netclaw.Actors/Sessions/SessionProtocol.Responses.cs b/src/Netclaw.Actors/Sessions/SessionProtocol.Responses.cs index 8d1d7a18b..1558ee16f 100644 --- a/src/Netclaw.Actors/Sessions/SessionProtocol.Responses.cs +++ b/src/Netclaw.Actors/Sessions/SessionProtocol.Responses.cs @@ -28,4 +28,13 @@ public sealed record CommandNack(SessionId SessionId, string Reason) : ISessionR public static CommandNack For(SessionId sessionId, string reason) => new(sessionId, reason); } + + /// + /// The session cannot accept the command now, but a later attempt can succeed. + /// + public sealed record CommandDeferred(SessionId SessionId, string Reason) : ISessionResponse + { + public static CommandDeferred For(SessionId sessionId, string reason) => + new(sessionId, reason); + } } diff --git a/src/Netclaw.Actors/Sessions/SessionProtocol.cs b/src/Netclaw.Actors/Sessions/SessionProtocol.cs index 760ca6d91..fe8cb921d 100644 --- a/src/Netclaw.Actors/Sessions/SessionProtocol.cs +++ b/src/Netclaw.Actors/Sessions/SessionProtocol.cs @@ -47,7 +47,8 @@ public interface ISessionQuery : IWithSessionId /// /// Marker for a reply the session actor sends in response to a command/query Ask — - /// (accepted) and (rejected). + /// (accepted), (retry later), + /// and (rejected). /// Lets callers (channel bindings, HTTP callback endpoints, the reminder execution /// actor) declare a typed Ask response instead of an untyped object. /// Transient: replies are local-dispatch only. diff --git a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs index c1d77dce8..384804874 100644 --- a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs +++ b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs @@ -70,11 +70,8 @@ internal sealed class DiscordSessionBindingActor : ReceivePersistentActor, IWith // was produced" — it suppresses the empty-turn fallback so a failed post // isn't followed by a misleading "I didn't manage to produce a reply". private bool _postFailedThisTurn; - // Reply targets for in-flight reminder delivery confirmations, keyed by - // reminder delivery key. Captured from DeliverTrustedSessionTurn; each is - // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. - // Keyed (not a single field) because multiple reminders can target the - // same session concurrently — a single field would be clobbered. + // Each key connects an accepted reminder turn to its delivery observer. + // The actor removes the entry after turn completion or enqueue failure. private readonly Dictionary _reminderDeliveryObservers = new(); private Netclaw.Actors.Protocol.TurnNumber _turnNumber; private string? _lastSetThreadName; @@ -1066,18 +1063,25 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) return; } + if (_turnInFlight) + { + _log.Info("Deferring CurrentSession reminder because the session has an active turn"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The target session has an active turn.")); + return; + } + if (_dependencies.IngressGate?.ClosedReason is { } ingressClosedReason) { - _log.Info("Rejecting Mode B reminder while restart drain is active"); - ackTarget.Tell(CommandNack.For(_sessionId, ingressClosedReason)); + _log.Info("Deferring CurrentSession reminder while restart drain is active"); + ackTarget.Tell(CommandDeferred.For(_sessionId, ingressClosedReason)); return; } var writer = _handle.InputQueue; if (writer is null) { - _log.Warning("Input queue is not initialized; rejecting Mode B reminder"); - ackTarget.Tell(CommandNack.For(_sessionId, "Discord session pipeline not initialized")); + _log.Warning("Input queue is not initialized; deferring CurrentSession reminder"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "Discord session pipeline is not initialized.")); return; } @@ -1098,10 +1102,8 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) AckTarget = ackTarget }; - // Only delivery-gated (DeliveryRequired) reminders carry a - // DeliveryObserver. Key it by the per-fire reminder delivery id so a - // second concurrent reminder to this session can't overwrite the - // first's observer before its turn reaches TurnCompleted. + // Only delivery-gated reminders carry a DeliveryObserver. + // The per-fire key connects the completed turn to the correct observer. if (message.Source.DeliveryObserver is { } deliveryObserver && message.Source.ReminderId is { } reminderKey && !string.IsNullOrWhiteSpace(reminderKey.Value)) @@ -1111,6 +1113,7 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) { using var writeCts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); await writer.WriteAsync(input, writeCts.Token); + _turnInFlight = true; _log.Debug( "reminder_mode_b_dispatch session={Session} reminder={Reminder}", _sessionId.Value, message.Source.ReminderId); @@ -1118,15 +1121,23 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) catch (OperationCanceledException) { _log.Warning("Timed out enqueueing Mode B reminder for session {0}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "Pipeline enqueue timeout")); + RemoveReminderDeliveryObserver(message.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The Discord pipeline enqueue timed out.")); } catch (ChannelClosedException) { - _log.Warning("Input queue closed; rejecting Mode B reminder for session {0}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "Pipeline input queue closed")); + _log.Warning("Input queue closed; deferring CurrentSession reminder for session {0}", _sessionId.Value); + RemoveReminderDeliveryObserver(message.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The Discord pipeline input queue is closed.")); } } + private void RemoveReminderDeliveryObserver(ReminderId? reminderKey) + { + if (reminderKey is { } key) + _reminderDeliveryObservers.Remove(key); + } + private enum ApprovalLookupResult { Matched, WrongRequester, NotFound } private ChannelDeliveryTargetInfo BuildDefaultDeliveryTarget() diff --git a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs index c6357d06e..e6a6ebefb 100644 --- a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs +++ b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs @@ -69,11 +69,8 @@ internal sealed class MattermostSessionBindingActor : ReceivePersistentActor, IW // was produced" — it suppresses the empty-turn fallback so a failed post // isn't followed by a misleading "I didn't manage to produce a reply". private bool _postFailedThisTurn; - // Reply targets for in-flight reminder delivery confirmations, keyed by - // reminder delivery key. Captured from DeliverTrustedSessionTurn; each is - // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. - // Keyed (not a single field) because multiple reminders can target the - // same session concurrently — a single field would be clobbered. + // Each key connects an accepted reminder turn to its delivery observer. + // The actor removes the entry after turn completion or enqueue failure. private readonly Dictionary _reminderDeliveryObservers = new(); private TurnNumber _turnNumber; private string? _cursorPostId; @@ -1044,18 +1041,25 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) return; } + if (_turnInFlight) + { + _log.Info("Deferring CurrentSession reminder because the session has an active turn"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The target session has an active turn.")); + return; + } + if (_dependencies.IngressGate?.ClosedReason is { } ingressClosedReason) { - _log.Info("Rejecting Mode B reminder while restart drain is active"); - ackTarget.Tell(CommandNack.For(_sessionId, ingressClosedReason)); + _log.Info("Deferring CurrentSession reminder while restart drain is active"); + ackTarget.Tell(CommandDeferred.For(_sessionId, ingressClosedReason)); return; } var writer = _handle.InputQueue; if (writer is null) { - _log.Warning("Input queue is not initialized; rejecting Mode B reminder"); - ackTarget.Tell(CommandNack.For(_sessionId, "Mattermost session pipeline not initialized")); + _log.Warning("Input queue is not initialized; deferring CurrentSession reminder"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "Mattermost session pipeline is not initialized.")); return; } @@ -1076,10 +1080,8 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) AckTarget = ackTarget }; - // Only delivery-gated (DeliveryRequired) reminders carry a - // DeliveryObserver. Key it by the per-fire reminder delivery id so a - // second concurrent reminder to this session can't overwrite the - // first's observer before its turn reaches TurnCompleted. + // Only delivery-gated reminders carry a DeliveryObserver. + // The per-fire key connects the completed turn to the correct observer. if (message.Source.DeliveryObserver is { } deliveryObserver && message.Source.ReminderId is { } reminderKey && !string.IsNullOrWhiteSpace(reminderKey.Value)) @@ -1089,6 +1091,7 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) { using var writeCts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); await writer.WriteAsync(input, writeCts.Token); + _turnInFlight = true; _log.Debug( "reminder_mode_b_dispatch session={Session} reminder={Reminder}", _sessionId.Value, message.Source.ReminderId); @@ -1096,15 +1099,23 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) catch (OperationCanceledException) { _log.Warning("Timed out enqueueing Mode B reminder for session {0}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "Pipeline enqueue timeout")); + RemoveReminderDeliveryObserver(message.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The Mattermost pipeline enqueue timed out.")); } catch (ChannelClosedException) { - _log.Warning("Input queue closed; rejecting Mode B reminder for session {0}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "Pipeline input queue closed")); + _log.Warning("Input queue closed; deferring CurrentSession reminder for session {0}", _sessionId.Value); + RemoveReminderDeliveryObserver(message.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The Mattermost pipeline input queue is closed.")); } } + private void RemoveReminderDeliveryObserver(ReminderId? reminderKey) + { + if (reminderKey is { } key) + _reminderDeliveryObservers.Remove(key); + } + private enum ApprovalLookupResult { Matched, WrongRequester, NotFound } private ChannelDeliveryTargetInfo BuildDefaultDeliveryTarget() diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index 7d8958f1a..aa80ef5f6 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -40,11 +40,8 @@ internal sealed class SlackThreadBindingActor : ReceivePersistentActor, IWithTim private bool _postedThisTurn; private bool _uploadedFileThisTurn; private PostResult? _lastFailedPost; - // Reply targets for in-flight reminder delivery confirmations, keyed by - // reminder delivery key. Captured from DeliverTrustedSessionTurn; each is - // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. - // Keyed (not a single field) because multiple reminders can target the - // same session concurrently — a single field would be clobbered. + // Each key connects an accepted reminder turn to its delivery observer. + // The actor removes the entry after turn completion or enqueue failure. private readonly Dictionary _reminderDeliveryObservers = new(); private readonly List _pendingApprovalRequests = []; @@ -260,18 +257,25 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) return; } + if (_turnInFlight) + { + _log.Info("Deferring CurrentSession reminder because the session has an active turn"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The target session has an active turn.")); + return; + } + if (_dependencies.IngressGate?.ClosedReason is { } ingressClosedReason) { - _log.Info("Rejecting Mode B reminder while restart drain is active"); - ackTarget.Tell(CommandNack.For(_sessionId, ingressClosedReason)); + _log.Info("Deferring CurrentSession reminder while restart drain is active"); + ackTarget.Tell(CommandDeferred.For(_sessionId, ingressClosedReason)); return; } var writer = _handle.InputQueue; if (writer is null) { - _log.Warning("Thread input queue is not initialized; rejecting Mode B reminder"); - ackTarget.Tell(CommandNack.For(_sessionId, "Slack thread pipeline not initialized")); + _log.Warning("Thread input queue is not initialized; deferring CurrentSession reminder"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "Slack thread pipeline is not initialized.")); return; } @@ -292,10 +296,8 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) AckTarget = ackTarget }; - // Only delivery-gated (DeliveryRequired) reminders carry a - // DeliveryObserver. Key it by the per-fire reminder delivery id so a - // second concurrent reminder to this session can't overwrite the - // first's observer before its turn reaches TurnCompleted. + // Only delivery-gated reminders carry a DeliveryObserver. + // The per-fire key connects the completed turn to the correct observer. if (message.Source.DeliveryObserver is { } deliveryObserver && message.Source.ReminderId is { } reminderKey && !string.IsNullOrWhiteSpace(reminderKey.Value)) @@ -305,6 +307,7 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) { using var writeCts = new CancellationTokenSource(OperationTimeout); await writer.WriteAsync(input, writeCts.Token); + _turnInFlight = true; _log.Debug( "reminder_mode_b_dispatch session={Session} reminder={Reminder}", _sessionId.Value, message.Source.ReminderId); @@ -312,15 +315,23 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) catch (OperationCanceledException) { _log.Warning("Timed out enqueueing Mode B reminder for session {0}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "Pipeline enqueue timeout")); + RemoveReminderDeliveryObserver(message.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The Slack pipeline enqueue timed out.")); } catch (ChannelClosedException) { - _log.Warning("Thread input queue closed; rejecting Mode B reminder for session {0}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "Pipeline input queue closed")); + _log.Warning("Thread input queue closed; deferring CurrentSession reminder for session {0}", _sessionId.Value); + RemoveReminderDeliveryObserver(message.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The Slack pipeline input queue is closed.")); } } + private void RemoveReminderDeliveryObserver(ReminderId? reminderKey) + { + if (reminderKey is { } key) + _reminderDeliveryObservers.Remove(key); + } + private async Task HandleInboundAsync(SlackThreadInbound message) { var inboundLog = _log diff --git a/src/Netclaw.Daemon.Tests/Gateway/SignalRSessionActorTests.cs b/src/Netclaw.Daemon.Tests/Gateway/SignalRSessionActorTests.cs new file mode 100644 index 000000000..b3b99f09a --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Gateway/SignalRSessionActorTests.cs @@ -0,0 +1,166 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using System.Collections.Concurrent; +using Akka; +using Akka.Actor; +using Akka.Hosting; +using Akka.Hosting.TestKit; +using Akka.Streams; +using Akka.Streams.Dsl; +using Microsoft.AspNetCore.SignalR; +using Netclaw.Actors.Channels; +using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; +using Netclaw.Configuration; +using Netclaw.Daemon.Gateway; +using Xunit; +using static Netclaw.Actors.Reminders.ReminderProtocol; +using static Netclaw.Actors.Sessions.SessionProtocol; + +namespace Netclaw.Daemon.Tests.Gateway; + +public sealed class SignalRSessionActorTests(ITestOutputHelper output) : TestKit(output: output) +{ + protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IServiceProvider provider) { } + + [Fact] + public async Task Busy_session_defers_a_CurrentSession_reminder() + { + var cancellationToken = TestContext.Current.CancellationToken; + var sessionId = new SessionId("signalr/busy-reminder-session"); + var pipeline = new CapturingSessionPipeline(); + var actor = Sys.ActorOf( + SignalRSessionActor.CreateProps(sessionId.Value, pipeline, new StubHubContext()), + "signalr-busy-reminder-session"); + + actor.Tell(new StartSignalRSession( + sessionId, + ChannelType.SignalR, + new SignalRConnectionId("connection-1"))); + await pipeline.Created.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken); + + actor.Tell(CreateReminderTurn(sessionId, "reminder-a")); + await AwaitAssertAsync( + () => Assert.Single(pipeline.Inputs), + cancellationToken: cancellationToken); + + var secondAttempt = CreateTestProbe(); + secondAttempt.Send(actor, CreateReminderTurn(sessionId, "reminder-b")); + var deferred = await secondAttempt.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: cancellationToken); + + Assert.Equal(sessionId, deferred.SessionId); + Assert.Contains("active turn", deferred.Reason, StringComparison.OrdinalIgnoreCase); + await AwaitAssertAsync( + () => Assert.Single(pipeline.Inputs), + cancellationToken: cancellationToken); + } + + private static DeliverTrustedSessionTurn CreateReminderTurn(SessionId sessionId, string reminderId) + { + var key = new ReminderId(reminderId); + return new DeliverTrustedSessionTurn( + sessionId, + "Run the reminder.", + new MessageSource + { + ChannelType = ChannelType.SignalR, + SenderId = new SenderId("reminder-system"), + Audience = TrustAudience.Team, + Boundary = TrustBoundary.Team, + Principal = PrincipalClassification.VerifiedAutomation, + Provenance = new SourceProvenance( + TransportAuthenticity.LocalProcess, + PayloadTaint.Trusted) + { + SourceKind = new SourceKind("reminder") + }, + ReceivedAt = DateTimeOffset.UnixEpoch, + ReminderId = key + }); + } + + private sealed class CapturingSessionPipeline : ISessionPipeline + { + private readonly TaskCompletionSource _created = new( + TaskCreationOptions.RunContinuationsAsynchronously); + + public Task Created => _created.Task; + + public ConcurrentQueue Inputs { get; } = new(); + + public Task CreateAsync( + SessionId sessionId, + SessionPipelineOptions options, + IMaterializer? materializer = null, + CancellationToken cancellationToken = default) + { + var killSwitch = KillSwitches.Shared($"signalr-test-{sessionId.Value}"); + var input = Sink.ForEach(Inputs.Enqueue).ObservingFault(); + var output = Source.Never().Via(killSwitch.Flow()); + _created.TrySetResult(); + return Task.FromResult(new MaterializedSession(input, output, killSwitch)); + } + + public Task SendFeedbackAsync(IWithSessionId feedback, CancellationToken ct = default) => + Task.CompletedTask; + + public Task SendFeedbackAndWaitAsync( + IWithSessionId feedback, + CancellationToken ct = default) => + Task.FromResult(CommandAck.For(feedback.SessionId)); + } + + private sealed class StubHubContext : IHubContext + { + public IHubClients Clients { get; } = new StubHubClients(); + + public IGroupManager Groups { get; } = new StubGroupManager(); + } + + private sealed class StubHubClients : IHubClients + { + private static readonly ISessionHubClient ClientInstance = new StubHubClient(); + + public ISessionHubClient All => ClientInstance; + + public ISessionHubClient AllExcept(IReadOnlyList excludedConnectionIds) => ClientInstance; + + public ISessionHubClient Client(string connectionId) => ClientInstance; + + public ISessionHubClient Clients(IReadOnlyList connectionIds) => ClientInstance; + + public ISessionHubClient Group(string groupName) => ClientInstance; + + public ISessionHubClient GroupExcept( + string groupName, + IReadOnlyList excludedConnectionIds) => ClientInstance; + + public ISessionHubClient Groups(IReadOnlyList groupNames) => ClientInstance; + + public ISessionHubClient User(string userId) => ClientInstance; + + public ISessionHubClient Users(IReadOnlyList userIds) => ClientInstance; + } + + private sealed class StubHubClient : ISessionHubClient + { + public Task ReceiveOutput(SessionOutputDto output) => Task.CompletedTask; + } + + private sealed class StubGroupManager : IGroupManager + { + public Task AddToGroupAsync( + string connectionId, + string groupName, + CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task RemoveFromGroupAsync( + string connectionId, + string groupName, + CancellationToken cancellationToken = default) => Task.CompletedTask; + } +} diff --git a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs index df4ca7450..035cf3505 100644 --- a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs +++ b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs @@ -33,11 +33,9 @@ internal sealed class SignalRSessionActor : ReceiveActor, IWithUnboundedStash, I private SignalRConnectionId _currentConnectionId; private Actors.Channels.ChannelType _channelType = Actors.Channels.ChannelType.Tui; private bool _deliveredThisTurn; - // Reply targets for in-flight reminder delivery confirmations, keyed by - // reminder delivery key. Captured from DeliverTrustedSessionTurn; each is - // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. - // Keyed (not a single field) because multiple reminders can target the - // same session concurrently — a single field would be clobbered. + private bool _turnInFlight; + // Each key connects an accepted reminder turn to its delivery observer. + // The actor removes the entry after turn completion or enqueue failure. private readonly Dictionary _reminderDeliveryObservers = new(); private static readonly TimeSpan PipelineInitTimeout = TimeSpan.FromSeconds(15); @@ -102,7 +100,7 @@ private void Initializing() catch (Exception ex) { _log.Error(ex, "Failed to initialize SignalR session pipeline for Mode B reminder; stopping actor"); - Sender.Tell(CommandNack.For(_sessionId, $"SignalR pipeline init failed: {ex.Message}")); + Sender.Tell(CommandDeferred.For(_sessionId, $"SignalR pipeline initialization failed: {ex.Message}")); Context.Stop(Self); } }); @@ -131,6 +129,7 @@ private void Active() { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); await writer.WriteAsync(msg.Input, cts.Token); + _turnInFlight = true; } catch (OperationCanceledException) { @@ -161,6 +160,7 @@ private void Active() ReceiveAsync(async msg => { _deliveredThisTurn = false; + _turnInFlight = false; // A reinit aborts any in-flight reminder turn before its // TurnCompleted. Report those as not-delivered now so the // execution actor redelivers immediately instead of stalling @@ -188,13 +188,20 @@ await _handle.ReinitializeAsync( { var ackTarget = Sender; + if (_turnInFlight) + { + _log.Info("Deferring CurrentSession reminder because the session has an active turn"); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The target session has an active turn.")); + return; + } + var writer = _handle.InputQueue; if (writer is null) { _log.Warning( - "SignalR input queue not initialized; rejecting Mode B reminder for {SessionId}", + "SignalR input queue not initialized; deferring CurrentSession reminder for {SessionId}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "SignalR pipeline not initialized")); + ackTarget.Tell(CommandDeferred.For(_sessionId, "SignalR pipeline is not initialized.")); return; } @@ -215,10 +222,8 @@ await _handle.ReinitializeAsync( AckTarget = ackTarget }; - // Only delivery-gated (DeliveryRequired) reminders carry a - // DeliveryObserver. Key it by the per-fire reminder delivery id so a - // second concurrent reminder to this session can't overwrite the - // first's observer before its turn reaches TurnCompleted. + // Only delivery-gated reminders carry a DeliveryObserver. + // The per-fire key connects the completed turn to the correct observer. if (msg.Source.DeliveryObserver is { } deliveryObserver && msg.Source.ReminderId is { } reminderKey && !string.IsNullOrWhiteSpace(reminderKey.Value)) @@ -228,6 +233,7 @@ await _handle.ReinitializeAsync( { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); await writer.WriteAsync(input, cts.Token); + _turnInFlight = true; _log.Debug( "reminder_mode_b_dispatch session={Session} reminder={Reminder}", _sessionId.Value, msg.Source.ReminderId); @@ -235,12 +241,14 @@ await _handle.ReinitializeAsync( catch (OperationCanceledException) { _log.Warning("Timed out enqueueing Mode B reminder for session {SessionId}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "SignalR pipeline enqueue timeout")); + RemoveReminderDeliveryObserver(msg.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The SignalR pipeline enqueue timed out.")); } catch (ChannelClosedException) { - _log.Warning("SignalR input queue closed; rejecting Mode B reminder for {SessionId}", _sessionId.Value); - ackTarget.Tell(CommandNack.For(_sessionId, "SignalR input queue closed")); + _log.Warning("SignalR input queue closed; deferring CurrentSession reminder for {SessionId}", _sessionId.Value); + RemoveReminderDeliveryObserver(msg.Source.ReminderId); + ackTarget.Tell(CommandDeferred.For(_sessionId, "The SignalR input queue is closed.")); Self.Tell(new ReinitializePipeline("input queue closed during Mode B delivery")); } }); @@ -275,6 +283,7 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) { ReportReminderDeliveryResult(noConnTurn, delivered: false); _deliveredThisTurn = false; + _turnInFlight = false; } return; } @@ -292,6 +301,7 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) { ReportReminderDeliveryResult(completed, delivered: _deliveredThisTurn); _deliveredThisTurn = false; + _turnInFlight = false; } } catch (Exception ex) @@ -305,6 +315,7 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) { ReportReminderDeliveryResult(failedTurn, delivered: false); _deliveredThisTurn = false; + _turnInFlight = false; } } } @@ -329,6 +340,12 @@ private void ReportReminderDeliveryResult(TurnCompleted completed, bool delivere } } + private void RemoveReminderDeliveryObserver(ReminderId? reminderKey) + { + if (reminderKey is { } key) + _reminderDeliveryObservers.Remove(key); + } + /// /// Tells every in-flight reminder observer that delivery did not happen, /// then clears them. Called when a turn can no longer reach TurnCompleted