PR 6.3 — Buffered update interval landing

Wires MxAccess.PublishingIntervalMs into the gw's SubscribeBulk
bufferedUpdateIntervalMs parameter on both subscribe paths:

- GalaxyDriver.SubscribeAsync — when the caller passes TimeSpan.Zero
  (typical for infrastructure callers like the deploy watcher), the
  driver substitutes _options.MxAccess.PublishingIntervalMs. When the
  caller sets a non-zero interval (the server's UA subscription
  publishingInterval), that wins.
- PerPlatformProbeWatcher — new bufferedUpdateIntervalMs ctor parameter
  defaulting to 0 (gw default cadence). GalaxyDriver passes
  _options.MxAccess.PublishingIntervalMs so probe ScanState changes
  publish at the configured rate.

Tests: caller-wins-when-non-zero, fallback-to-config-when-zero on the
driver; default-zero, configured-forwarded, negative-rejected on the
probe watcher.

A session-level SetBufferedUpdateInterval RPC exists in the gw protocol
(MxCommandKind.SetBufferedUpdateInterval) but the .NET client doesn't
expose a typed helper yet — adjusting an existing subscription's
interval is a follow-up. Today's path subscribes once with the right
interval, which covers the common case.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-04-29 16:56:33 -04:00
parent 7b21c3b428
commit 2fdad81af3
4 changed files with 101 additions and 6 deletions

View File

@@ -209,7 +209,9 @@ public sealed class GalaxyDriver
_supervisor.StateChanged += OnSupervisorStateChanged;
_probeWatcher = new PerPlatformProbeWatcher(_subscriber, _hostStatuses, _logger);
_probeWatcher = new PerPlatformProbeWatcher(
_subscriber, _hostStatuses, _logger,
bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs);
}
/// <summary>
@@ -441,7 +443,12 @@ public sealed class GalaxyDriver
return new GalaxySubscriptionHandle(subscriptionId);
}
var bufferedIntervalMs = (int)Math.Max(0, publishingInterval.TotalMilliseconds);
// PR 6.3 — when the caller doesn't set a publishing interval (TimeSpan.Zero or
// negative), fall back to the configured MxAccess.PublishingIntervalMs. The
// server's UA subscription publishingInterval drives this in production; tests
// and infrastructure callers (probe watcher, deploy watcher) hit the fallback.
var requested = (int)Math.Max(0, publishingInterval.TotalMilliseconds);
var bufferedIntervalMs = requested > 0 ? requested : _options.MxAccess.PublishingIntervalMs;
var results = await _subscriber
.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, cancellationToken)
.ConfigureAwait(false);

View File

@@ -36,6 +36,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable
private readonly IGalaxySubscriber _subscriber;
private readonly HostStatusAggregator _aggregator;
private readonly ILogger _logger;
private readonly int _bufferedUpdateIntervalMs;
// Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe;
// we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call.
@@ -45,11 +46,20 @@ public sealed class PerPlatformProbeWatcher : IDisposable
private bool _disposed;
public PerPlatformProbeWatcher(
IGalaxySubscriber subscriber, HostStatusAggregator aggregator, ILogger? logger = null)
IGalaxySubscriber subscriber,
HostStatusAggregator aggregator,
ILogger? logger = null,
int bufferedUpdateIntervalMs = 0)
{
_subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
_aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator));
_logger = logger ?? NullLogger.Instance;
if (bufferedUpdateIntervalMs < 0)
{
throw new ArgumentOutOfRangeException(nameof(bufferedUpdateIntervalMs),
"bufferedUpdateIntervalMs must be >= 0; 0 means use the gw's default cadence.");
}
_bufferedUpdateIntervalMs = bufferedUpdateIntervalMs;
}
/// <summary>Snapshot of platform tag names currently watched.</summary>
@@ -107,10 +117,12 @@ public sealed class PerPlatformProbeWatcher : IDisposable
if (toAdd.Count == 0) return;
var probeAddresses = toAdd.Select(p => p + ProbeSuffix).ToArray();
// bufferedUpdateInterval=0 — probe ScanState changes are rare enough that the gw's
// default cadence is fine; explicit polling rate goes through PR 6.3.
// PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
// when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
// changes are rare so a coarser interval is usually fine; deployments that need
// tighter health visibility can dial it down through GalaxyDriverOptions.
var results = await _subscriber.SubscribeBulkAsync(
probeAddresses, bufferedUpdateIntervalMs: 0, cancellationToken).ConfigureAwait(false);
probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false);
for (var i = 0; i < toAdd.Count; i++)
{