Native messaging integration with shared job tracking

.agents/skills/foundatio/SKILL.md

@@ -358,3 +358,11 @@ Validate a custom transport or job store against the shared conformance suites i
 | `Foundatio.Xunit` | xUnit v2 test logging, retry attributes |
 | `Foundatio.Xunit.v3` | xUnit v3 test logging, retry attributes |
 | `Foundatio.DataProtection` | ASP.NET Core Data Protection key storage via `IFileStorage` |
+
+## Broker execution integration
+
+- Optional broker history uses the normal job store: `.Jobs.UseInMemory()` / `.Jobs.UseRedis()`, `IJobRuntimeStore`, `JobState`, and `IJobMonitor`. Create records with `ExecutionOwner = Broker`; runtime claims and recovery exclude them. `BeginBrokerAttemptAsync` returns a fresh claim token for `ReportJobProgressAsync` / `CompleteJobAsync`. Broker delivery leases stay in the message bus. `MessageExecutionPipeline` and `MessageProcessingContext` connect processing to the shared job store without scheduling the same work twice.
+- `ConsumeWithOutcomeAsync` handles returned Success/Retry/DeadLetter/Unsettled outcomes. Endpoint receive capacity, visibility, automatic renewal, and graceful drain are configured with MessageHandlerOptions.
+- `SubscribeNodeAsync` supplies independent best-effort node broadcasts. AWS uses managed tagged resources and startup stale cleanup, not native TTL. Acknowledge-before-callback intentionally allows lost notifications.
+- Native `MessageAdministration` owns bounded dead-letter inspection/replay. Send-before-delete is at least once, not atomic. Optional replay preparation can create fresh tracked execution IDs.
+- `.Locking.UseRedis()` uses native RedisLockProvider with ownership-checked renewal/release. It coordinates live resource ownership, not persistent duplicate detection.

docs/guide/locks.md

@@ -676,3 +676,7 @@ if (lck is not null)
 - [Jobs](./jobs) - Background jobs with distributed locking
 - [Resilience](./resilience) - Retry policies for lock acquisition
 - [Serialization](./serialization) - Serializer configuration and performance
+
+## Native Redis resource locks
+
+`foundatio.Locking.UseRedis()` registers `RedisLockProvider` using the application's shared `IConnectionMultiplexer`. Acquisition uses Redis `SET NX` with expiration; renewal and release compare the ownership token atomically. `LockOwnershipLostException` signals that a holder may no longer extend its lease. Expired holders cannot release a successor's lock. These are leased resource locks, not deduplication or fencing of external writes. Configure distinct key prefixes for unrelated applications.

docs/guide/messaging.md

@@ -153,3 +153,25 @@ SQS/SNS support varies by destination role. Do not infer topic capabilities from
 ## Migration
 
 The former publish-only interfaces live under `Foundatio.Messaging.Legacy`; `Messaging.AddLegacyAdapter()` adapts them to this bus. Legacy `IQueue<T>` workers migrate to an explicit consumer and `SendAsync`. There is no `Deliveries.Both`, handler-name-derived subscriber identity, or `PerInstance` flag. Choose `AddConsumer` or `AddSubscriber`, and use a stable service identity or explicit names for durable subscribers.
+
+## Broker-driven execution tracking
+
+For queues that need progress, cancellation, and operational history, configure `.Jobs.UseInMemory()` or `.Jobs.UseRedis()`. Tracked deliveries use the existing `IJobRuntimeStore`, `JobState`, and `IJobMonitor`. Set `ExecutionOwner = JobExecutionOwner.Broker` when recording enqueue acceptance: the ordinary job worker cannot claim or recover these records. The message bus owns delivery leases and settlement, while each delivery attempt gets a fresh token for atomic progress and completion updates. Registering the store starts no job worker or scheduler.
+
+`MessageExecutionPipeline` runs a raw delivery callback with `MessageProcessingContext`, applies returned `MessageOutcome` values, and persists only confirmed settlement. Pass it to a manual-ack raw `ConsumeAsync` endpoint with `WaitForManualSettlement = false`. The producer creates a unique execution state before sending and puts its ID in `ExecutionHeaders.ExecutionId`. Integrations such as Foundatio.Mediator perform this composition automatically.
+
+Processing claims and subsequent progress/status writes are fenced by the broker attempt number. A terminal execution is not started again. Expired history does not block valid broker work; progress/history updates then become no-ops. Tracking is not persistent deduplication or exactly-once delivery. Broker acceptance and persistence are separate operations; uncertain sends and acknowledgments remain nonterminal. Application side effects must tolerate retries.
+
+Endpoint options own `MaxConcurrency`, `PrefetchCount`, `VisibilityTimeout`, `AutoRenewLock`, and `ShutdownTimeout`. Manual and automatic renewal share one conservative delivery deadline. Lost ownership cancels processing. Stopping receiving allows admitted handlers to drain within the configured shutdown timeout.
+
+`ConsumeWithOutcomeAsync` accepts `MessageOutcome.Success`, `Retry`, `DeadLetter`, or `Unsettled` without requiring expected application failures to throw exceptions.
+
+## Per-node broadcasts
+
+`SubscribeNodeAsync` receives an independent best-effort copy on each running node, acknowledging before callbacks. Memory and Redis use native expiring subscriptions. AWS creates a tagged SQS queue and SNS subscription, heartbeats ownership, cleans up on disposal, and reaps stale resources when another node starts. AWS managed resources do not advertise native TTL expiration. Configure IAM for the required resource lifecycle, tagging, and discovery operations, including when durable topology is externally provisioned.
+
+Use this for invalidation and UI refresh signals. A disconnected, paused, or restarting node may miss events and should refresh authoritative state. Use durable service subscriptions for work that needs replay and retries.
+
+## Operational recovery
+
+`MessageAdministration` supplies statistics, bounded dead-letter inspection, deletion, and replay through native capabilities. Providers without native inspection use bounded receive/hold/release. Scans stop at 1,000 messages or ten seconds; their cleanup has an independent timeout. A successful replacement send precedes deleting the original. That fallback is not transactional and an uncertain send can produce a duplicate. Callers may prepare fresh execution metadata for an operator replay while retaining the failed history.

src/Foundatio.Aws/AwsFoundatioBuilderExtensions.cs

@@ -19,7 +19,7 @@ public static FoundatioBuilder.MessagingBuilder UseAws(this FoundatioBuilder.Mes
     {
         return builder.UseTransport(sp =>
         {
-            var options = new AwsMessageTransportOptions();
+            var options = new AwsMessageTransportOptions { LoggerFactory = sp.GetService<Microsoft.Extensions.Logging.ILoggerFactory>() };
             BindFromConfiguration(options, sp.GetService<IConfiguration>()?.GetSection("Aws"));
             configure?.Invoke(options);
             return new AwsMessageTransport(options);

src/Foundatio.Aws/AwsMessageTransport.Administration.cs

@@ -0,0 +1,60 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Amazon.SQS.Model;
+
+namespace Foundatio.Messaging;
+
+public sealed partial class AwsMessageTransport
+{
+    /// <summary>Validates existence and declared SQS attributes, reporting all mismatches together.</summary>
+    public async Task ValidateAsync(IReadOnlyList<DestinationDeclaration> declarations, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(declarations);
+        var problems = new List<string>();
+        foreach (var declaration in declarations)
+        {
+            if (!await ExistsAsync(declaration.Address, cancellationToken).ConfigureAwait(false))
+            {
+                problems.Add($"Destination '{declaration.Address.Key}' does not exist.");
+                continue;
+            }
+            if (declaration.Address.Role != DestinationRole.Queue) continue;
+            ValidateQueueArguments(declaration.ProviderArguments);
+            string url = await ResolveQueueUrlAsync(declaration.Address, cancellationToken).ConfigureAwait(false);
+            var response = await _sqs.Value.GetQueueAttributesAsync(new GetQueueAttributesRequest
+            {
+                QueueUrl = url,
+                AttributeNames = ["All"]
+            }, cancellationToken).ConfigureAwait(false);
+            foreach (var expected in declaration.ProviderArguments ?? new Dictionary<string, string>())
+            {
+                response.Attributes.TryGetValue(expected.Key, out var actual);
+                if (!String.Equals(expected.Value, actual, StringComparison.Ordinal))
+                    problems.Add($"Queue '{declaration.Address.Name}' attribute {expected.Key} is '{actual ?? "(unset)"}', expected '{expected.Value}'.");
+            }
+            if (response.Attributes.TryGetValue("RedrivePolicy", out string? redrive) && !String.IsNullOrEmpty(redrive))
+                problems.Add($"Queue '{declaration.Address.Name}' has a broker RedrivePolicy. Foundatio owns delivery attempts; remove the broker policy.");
+        }
+        if (problems.Count > 0) throw new InvalidOperationException(String.Join(Environment.NewLine, problems));
+    }
+
+    private static void ValidateQueueArguments(IReadOnlyDictionary<string, string>? arguments)
+    {
+        if (arguments is null) return;
+        if (arguments.TryGetValue("RedrivePolicy", out var policy) && !String.IsNullOrEmpty(policy))
+            throw new ArgumentException("Foundatio owns retry and dead-letter policy; do not configure a broker RedrivePolicy.", nameof(arguments));
+        foreach (var argument in arguments)
+        {
+            if (argument.Key is not ("VisibilityTimeout" or "MessageRetentionPeriod")) continue;
+            int min = argument.Key == "VisibilityTimeout" ? 1 : 60;
+            int max = argument.Key == "VisibilityTimeout" ? 43200 : 1209600;
+            if (!Int32.TryParse(argument.Value, NumberStyles.Integer, CultureInfo.InvariantCulture, out int value) || value < min || value > max)
+                throw new ArgumentException($"SQS {argument.Key} must be between {min} and {max} seconds.", nameof(arguments));
+        }
+    }
+
+}

src/Foundatio.Aws/AwsMessageTransport.Batching.cs

@@ -48,9 +48,9 @@ public async Task<SendResult> SendAsync(DestinationAddress destination, IReadOnl
         string address = topic ? await ResolveTopicArnAsync(destination.Name, ct).ConfigureAwait(false) : await ResolveQueueUrlAsync(destination, ct).ConfigureAwait(false);
         for (int offset = 0; offset < prepared.Count;)
         {
-            var batch = new List<PreparedMessage>(10);
+            var batch = new List<PreparedMessage>(_options.MaxBatchSize);
             int bytes = 0;
-            while (offset < prepared.Count && batch.Count < 10 && bytes + prepared[offset].Bytes <= maximumBytes)
+            while (offset < prepared.Count && batch.Count < _options.MaxBatchSize && bytes + prepared[offset].Bytes <= maximumBytes)
             {
                 var entry = prepared[offset++];
                 batch.Add(entry);

src/Foundatio.Aws/AwsMessageTransport.NodeSubscriptions.cs

@@ -0,0 +1,141 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Threading;
+using System.Threading.Tasks;
+using Amazon.SQS.Model;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace Foundatio.Messaging;
+
+public sealed partial class AwsMessageTransport
+{
+    private const string NodeRoleTag = "fnd:role";
+    private const string NodeTopicTag = "fnd:topic";
+    private const string NodeHeartbeatTag = "fnd:heartbeat";
+
+    /// <summary>Owns a tagged SQS subscription for one node; active nodes renew it and startup reaps stale peers.</summary>
+    public async Task<IManagedNodeSubscription> OpenNodeSubscriptionAsync(MessageNodeSubscriptionOptions options, CancellationToken cancellationToken = default)
+    {
+        options.Validate();
+        ArgumentOutOfRangeException.ThrowIfLessThan(options.MessageRetention, TimeSpan.FromMinutes(1));
+        ArgumentOutOfRangeException.ThrowIfGreaterThan(options.MessageRetention, TimeSpan.FromDays(14));
+        // A fresh identity avoids queue-deleted-recently failures and prevents two processes adopting one receipt namespace.
+        var source = DestinationAddress.ForSubscription(options.Topic, $"node-{options.NodeId}-{Guid.NewGuid():N}");
+        await EnsureSubscriptionAsync(source, cancellationToken).ConfigureAwait(false);
+        string url = await ResolveQueueUrlAsync(source, cancellationToken).ConfigureAwait(false);
+        try
+        {
+            await _sqs.Value.SetQueueAttributesAsync(new SetQueueAttributesRequest
+            {
+                QueueUrl = url,
+                Attributes = new Dictionary<string, string>
+                {
+                    ["MessageRetentionPeriod"] = Math.Ceiling(options.MessageRetention.TotalSeconds).ToString(CultureInfo.InvariantCulture)
+                }
+            }, cancellationToken).ConfigureAwait(false);
+            await TagNodeAsync(url, options.Topic, cancellationToken).ConfigureAwait(false);
+            await SweepNodesAsync(options.Topic, options.StaleAfter, cancellationToken).ConfigureAwait(false);
+            return new NodeOwner(this, source, url, options);
+        }
+        catch
+        {
+            using var cleanup = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+            try { await DeleteAsync(source, cleanup.Token).ConfigureAwait(false); } catch { }
+            throw;
+        }
+    }
+
+    private Task TagNodeAsync(string url, string topic, CancellationToken cancellationToken)
+        => _sqs.Value.TagQueueAsync(new TagQueueRequest
+        {
+            QueueUrl = url,
+            Tags = new Dictionary<string, string>
+            {
+                [NodeRoleTag] = "node-subscription",
+                [NodeTopicTag] = topic,
+                [NodeHeartbeatTag] = DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture)
+            }
+        }, cancellationToken);
+
+    private async Task SweepNodesAsync(string topic, TimeSpan staleAfter, CancellationToken cancellationToken)
+    {
+        string physicalPrefix = SanitizeResourceName(_options.ResourcePrefix + topic + "/node-");
+        physicalPrefix = physicalPrefix[..Math.Min(32, physicalPrefix.Length)];
+        string? cursor = null;
+        do
+        {
+            var page = await _sqs.Value.ListQueuesAsync(new ListQueuesRequest
+            {
+                QueueNamePrefix = physicalPrefix,
+                MaxResults = 100,
+                NextToken = cursor
+            }, cancellationToken).ConfigureAwait(false);
+            foreach (string url in page.QueueUrls ?? [])
+            {
+                try
+                {
+                    var tags = (await _sqs.Value.ListQueueTagsAsync(new ListQueueTagsRequest { QueueUrl = url }, cancellationToken).ConfigureAwait(false)).Tags;
+                    if (tags is null || !tags.TryGetValue(NodeRoleTag, out var role) || role != "node-subscription"
+                        || !tags.TryGetValue(NodeTopicTag, out var ownedTopic) || ownedTopic != topic
+                        || !tags.TryGetValue(NodeHeartbeatTag, out var heartbeat)
+                        || !DateTimeOffset.TryParse(heartbeat, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var lastSeen)
+                        || lastSeen >= DateTimeOffset.UtcNow - staleAfter) continue;
+                    // Re-read ownership immediately before deletion; a peer may have resumed during the scan.
+                    var current = (await _sqs.Value.ListQueueTagsAsync(new ListQueueTagsRequest { QueueUrl = url }, cancellationToken).ConfigureAwait(false)).Tags;
+                    if (current is null || !current.TryGetValue(NodeHeartbeatTag, out var value) || value != heartbeat) continue;
+                    var attributes = await _sqs.Value.GetQueueAttributesAsync(new GetQueueAttributesRequest { QueueUrl = url, AttributeNames = ["QueueArn"] }, cancellationToken).ConfigureAwait(false);
+                    string? topicArn = await FindTopicArnAsync(topic, cancellationToken).ConfigureAwait(false);
+                    if (topicArn is not null && await FindSubscriptionArnAsync(topicArn, attributes.QueueARN, cancellationToken).ConfigureAwait(false) is { } arn)
+                        await _sns.Value.UnsubscribeAsync(arn, cancellationToken).ConfigureAwait(false);
+                    await _sqs.Value.DeleteQueueAsync(url, cancellationToken).ConfigureAwait(false);
+                }
+                catch (QueueDoesNotExistException) { }
+            }
+            cursor = page.NextToken;
+        } while (!String.IsNullOrEmpty(cursor));
+    }
+
+    private sealed class NodeOwner : IManagedNodeSubscription
+    {
+        private readonly AwsMessageTransport _owner;
+        private readonly CancellationTokenSource _stop = new();
+        private readonly Task _heartbeat;
+        private readonly ILogger _logger;
+        private int _disposed;
+        public DestinationAddress Source { get; }
+
+        public NodeOwner(AwsMessageTransport owner, DestinationAddress source, string url, MessageNodeSubscriptionOptions options)
+        {
+            _owner = owner;
+            _logger = owner._options.LoggerFactory?.CreateLogger<AwsMessageTransport>() ?? NullLogger<AwsMessageTransport>.Instance;
+            Source = source;
+            _heartbeat = HeartbeatAsync(url, options);
+        }
+
+        private async Task HeartbeatAsync(string url, MessageNodeSubscriptionOptions options)
+        {
+            while (!_stop.IsCancellationRequested)
+            {
+                try
+                {
+                    await Task.Delay(options.HeartbeatInterval, _stop.Token).ConfigureAwait(false);
+                    await _owner.TagNodeAsync(url, options.Topic, _stop.Token).ConfigureAwait(false);
+                }
+                catch (OperationCanceledException) when (_stop.IsCancellationRequested) { return; }
+                catch (Exception exception) { _logger.LogWarning(exception, "Unable to renew node subscription {Source}; retrying on the next heartbeat", Source); }
+            }
+        }
+
+        public async ValueTask DisposeAsync()
+        {
+            if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+            await _stop.CancelAsync().ConfigureAwait(false);
+            await _heartbeat.ConfigureAwait(false);
+            using var cleanup = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+            try { await _owner.DeleteAsync(Source, cleanup.Token).ConfigureAwait(false); }
+            finally { _stop.Dispose(); }
+        }
+    }
+}