fix: yield synchronous hub drains to queued peers

[rbuergi]

A hub whose synchronous handler keeps posting another turn owns its scheduler worker until the queue empties. PreferFairness makes queued peer drains reachable but cannot preempt that running drain; a peer's queued shutdown therefore waits behind the entire chain.

The drain now yields after 64 synchronous turns, retaining its serialization latch and scheduling the next batch on the original scheduler. FIFO processing, the side lane, genuinely asynchronous turns, and scheduling-failure handling remain unchanged. The owning disposal-diagnostics documentation records the mechanism and its attribution limits.

Related: #3593, #4847. This fixes a deterministic starvation mechanism; it does not by itself attribute every historical production stall in those issues.

Validation:

• New real-hub regression failed before the fix: the peer reached quiescing only after all 1,024 hot-hub turns. The same regression checks early shutdown, FIFO delivery, and the configured scheduler after the fix.
• Full MeshWeaver.Messaging.Hub.Test: 494 passed, 0 failed, 0 skipped (3m18s), with a real TRX.
• Messaging test project and Documentation dependency graph built Release with -warnaserror: 0 warnings, 0 errors.

Production baseline reproduced on memex.meshweaver.cloud at 2026-09-22 05:31:20Z, framework 026442fff0f2728e96bb838494559b68b5c2b797: turns=1024 peerAtTurn=1024 fifo=True originalScheduler=True. Activity: rbuergi/_Activity/c2a91f2eb521417489ade5b4f6823647. The same Code node, rbuergi/Script/pump-fairness-20260922, will be executed unchanged after rollout; acceptance is 0 < peerAtTurn < 1024, all 1,024 turns in FIFO order, and the original scheduler retained. It creates only two ephemeral hubs and completes their disposal; it changes no global thread-pool limits or persistent business nodes.

This is host framework code, so rollout replaces the implementation; no individual node recycle address is required.

[rbuergi]

Production baseline, 2026-09-22 05:31:20Z, memex.meshweaver.cloud:

PUMP_FAIRNESS_PROBE turns=1024 peerAtTurn=1024 fifo=True originalScheduler=True framework=3.0.0+026442fff0f2728e96bb838494559b68b5c2b797

Activity rbuergi/_Activity/c2a91f2eb521417489ade5b4f6823647 completed Succeeded. This demonstrates the deployed drain's unbounded ownership of this shared scheduler worker, independently of the local regression. After deployment the exact same source must give 0 < peerAtTurn < 1024, with turns=1024, fifo=True, and originalScheduler=True.

Exact bounded production probe (preserved for repeatability)

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
using MeshWeaver.Messaging;
using MeshWeaver.Mesh;
using Microsoft.Extensions.Logging;

const int turnCount = 1024;
var pair = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, 1, 1);
var scheduler = pair.ExclusiveScheduler;
var processed = new ConcurrentQueue<int>();
var finished = new AsyncSubject<Unit>();
var atQuiescing = -1;
var wrongScheduler = 0;
var suffix = Guid.NewGuid().ToString("N");
var peer = (MessageHub)Mesh.GetHostedHub(new Address("peer-" + suffix, "pump-fairness-probe"), c => c
    .WithTaskScheduler(scheduler).WithPostingIdentity(PostingIdentity.System), HostedHubCreation.Always);
var busy = (MessageHub)Mesh.GetHostedHub(new Address("busy-" + suffix, "pump-fairness-probe"), c => c
    .WithTaskScheduler(scheduler).WithPostingIdentity(PostingIdentity.System)
    .WithHandler<int>((h, delivery) =>
    {
        if (TaskScheduler.Current != scheduler) Interlocked.Increment(ref wrongScheduler);
        processed.Enqueue(delivery.Message);
        if (delivery.Message == 0) peer.Dispose();
        if (delivery.Message + 1 < turnCount)
            h.Post(delivery.Message + 1, o => o.WithTarget(h.Address));
        else { finished.OnNext(Unit.Default); finished.OnCompleted(); }
        return delivery.Processed();
    }), HostedHubCreation.Always);
try
{
    await peer.RunLevelChanged.Where(l => l == MessageHubRunLevel.Started).Take(1)
        .Timeout(TimeSpan.FromSeconds(10)).Await(Ct);
    await busy.RunLevelChanged.Where(l => l == MessageHubRunLevel.Started).Take(1)
        .Timeout(TimeSpan.FromSeconds(10)).Await(Ct);
    using var observed = peer.RunLevelChanged.Where(l => l >= MessageHubRunLevel.Quiescing).Take(1)
        .Subscribe(_ => Volatile.Write(ref atQuiescing, processed.Count));
    busy.Post(0, o => o.WithTarget(busy.Address));
    await finished.Timeout(TimeSpan.FromSeconds(10)).Await(Ct);
    await peer.DisposalCompleted.Timeout(TimeSpan.FromSeconds(10)).Await(Ct);
    Log.LogInformation("PUMP_FAIRNESS_PROBE turns={Turns} peerAtTurn={PeerAtTurn} fifo={Fifo} originalScheduler={OriginalScheduler} framework={Framework}",
        processed.Count, atQuiescing, processed.SequenceEqual(Enumerable.Range(0, turnCount)), wrongScheduler == 0,
        typeof(MessageHub).Assembly.GetCustomAttributes(typeof(System.Reflection.AssemblyInformationalVersionAttribute), false)
            .Cast<System.Reflection.AssemblyInformationalVersionAttribute>().Single().InformationalVersion);
}
finally
{
    busy.Dispose(); peer.Dispose();
    await busy.DisposalCompleted.Timeout(TimeSpan.FromSeconds(10)).Await(Ct);
    await peer.DisposalCompleted.Timeout(TimeSpan.FromSeconds(10)).Await(Ct);
    pair.Complete();
    await pair.Completion;
    finished.Dispose();
}

[Copilot]

Copilot review overview

🟡 Changes recommended

The drain handoff has unresolved scheduling-failure and overlapping-drain race risks.

Get a fresh assessment by requesting another Copilot review.

Review effort: Lite
Findings: 1

Open (1)

• Early continuation scheduling breaks single-drain serialization · New

What changed in this PR

This PR bounds synchronous hub drain batches so queued peer hubs can progress during self-posting workloads.

Changes:

• Limits synchronous drains to 64 turns before rescheduling.
• Adds real-hub fairness, FIFO, and scheduler regression coverage.
• Documents the starvation mechanism and attribution limits.

File	Description
test/​MeshWeaver.Messaging.Hub.Test/​PumpDrainFairnessTest.cs	Adds fairness and FIFO regression coverage.
src/​MeshWeaver.Messaging.Hub/​MessageService.cs	Implements bounded drain scheduling.
src/​MeshWeaver.Documentation/​Data/​Architecture/​DisposalStallVerdicts.md	Documents disposal-stall diagnostics.

---

💡 Configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[github-actions]

Test Results (shard 1)

533 tests   533 ✅  58s ⏱️
  1 suites    0 💤
  1 files      0 ❌

Results for commit 358ec85.

[github-actions]

Test Results (shard 0)

  1 files    1 suites   3m 1s ⏱️
314 tests 314 ✅ 0 💤 0 ❌
318 runs  318 ✅ 0 💤 0 ❌

Results for commit 358ec85.

[github-actions]

Test Results (shard 3)

429 tests   429 ✅  1m 3s ⏱️
  2 suites    0 💤
  2 files      0 ❌

Results for commit 358ec85.

[rbuergi]

The same byte-identical bounded probe also reproduced the baseline on memex.systemorph.com at 2026-09-22 05:33:40Z:

PUMP_FAIRNESS_PROBE turns=1024 peerAtTurn=1024 fifo=True originalScheduler=True framework=3.0.0+8a52945608611948580ce2b472ede1993e11b43c

Activity rbuergi/_Activity/09ff3437531f4beb8e71e81dd9052cfc completed Succeeded; Code path rbuergi/pump-fairness-20260922. Both production baselines now exist, for their respective framework commits. The ordinary node read sometimes returned Unavailable; the activity's /data/ resource returned the positive result above, so no conclusion depends on missing logs or a timeout.

[github-actions]

Test Results (shard 4)

    2 files      2 suites   6m 9s ⏱️
2 529 tests 2 529 ✅ 0 💤 0 ❌
2 530 runs  2 530 ✅ 0 💤 0 ❌

Results for commit 358ec85.

[github-actions]

Test Results (shard 2)

1 551 tests   1 551 ✅  7m 37s ⏱️
    2 suites      0 💤
    2 files        0 ❌

Results for commit 358ec85.

[github-actions]

Test Results (shard 5)

    5 files      5 suites   15m 6s ⏱️
3 272 tests 3 272 ✅ 0 💤 0 ❌
3 276 runs  3 276 ✅ 0 💤 0 ❌

Results for commit 358ec85.

[github-actions]

Test Results

   13 files     13 suites   33m 55s ⏱️
8 628 tests 8 628 ✅ 0 💤 0 ❌
8 637 runs  8 637 ✅ 0 💤 0 ❌

Results for commit 358ec85.

[rbuergi]

Production verification passed on memex.systemorph.com, 3.0.0-ci.9162 / core c55301f396ef60e1a8a965334a407e433d97c553 (contains this PR), after status v2406 confirmed one converged generation with exactly two ready replicas.

The unchanged baseline script rbuergi/pump-fairness-20260922 ran once. Activity rbuergi/_Activity/0ed7b70c60de4e18a230d950d0454d7d Succeeded at 2026-09-22 09:56:29 UTC:

PUMP_FAIRNESS_PROBE turns=1024 peerAtTurn=64 fifo=True originalScheduler=True framework=3.0.0+c55301f396ef60e1a8a965334a407e433d97c553

Before → after: peer disposal was processed at turn 1024 → 64, with all 1,024 messages still FIFO and the original scheduler retained. Strict checks passed, including 0 < peerAtTurn < 1024. Current source was byte-identical to the original saved probe (2,933 characters; baseline hash VwH54i11SHaIdtvBEjlHOfZCINGR98Ly7ht/QvxwFPg=).

This proves the synchronous-drain fairness correction in production. Attribution of every broader historical #3593/#4847 stall remains separate; those issues are not closed by this verification.