The error occurs intermittently because of a race condition on the m_lastPipe field

[balazs-tancsin-ppe]

The Problem 🔴

The error occurs intermittently because of a race condition on the m_lastPipe field:

1. Poller receives subscription message → m_lastPipe is set to the subscriber's pipe (XPub.cs:398)
2. OnPubSocketOnReceiveReady event fires and your code calls Subscribe(topic)
3. BUT: While you're in Subscribe(), another subscription message arrives on a different subscriber
4. Poller processes it → calls XRecv() again → sets m_lastPipe to the NEW pipe
5. Then clears it → m_lastPipe = null (XPub.cs:271)
6. Your Subscribe() call checks → if (m_manual && m_lastPipe != null) → fails because it's now null
7. Falls through to generic Options.SetSocketOption() which doesn't know about Subscribe
8. Throws the error!

Why It's Intermittent 🎲

• High message rate: More likely to have multiple messages queued
• System load: Affects thread scheduling and timing
• Pure timing: Race condition depends on which thread runs first

Key Code Locations:

┌─────────────────┬─────────┬─────────────────────────────────────────────────────────────┐
│ File │ Lines │ Issue │
├─────────────────┼─────────┼─────────────────────────────────────────────────────────────┤
│ XPub.cs │ 265-274 │ Subscribe handler - requires m_lastPipe to be non-null │
├─────────────────┼─────────┼─────────────────────────────────────────────────────────────┤
│ XPub.cs │ 271 │ m_lastPipe = null; - race condition here │
├─────────────────┼─────────┼─────────────────────────────────────────────────────────────┤
│ XPub.cs │ 398 │ XRecv() sets m_lastPipe from new messages │
├─────────────────┼─────────┼─────────────────────────────────────────────────────────────┤
│ SocketBase.cs │ 336-340 │ SetSocketOption calls XSetSocketOption, falls back if false │
├─────────────────┼─────────┼─────────────────────────────────────────────────────────────┤
│ Options.cs │ 582 │ Throws error when Subscribe reaches here │
└─────────────────┴─────────┴─────────────────────────────────────────────────────────────┘

Evidence in the Code:

Even the test file acknowledges this (XPubSubTests.cs:479):

// NB Identity must be set before pub.Subscribe/Unsubscribe/Send,
// because these operations clear a private field with last subscriber

Test Objectives:

• Simulates random connect/disconnect timing between EventPublisher and 2 subscribers (ALPHA, BETA)
• Verifies EventPublisher can handle various startup orderings
• Validates that clients receive messages even if they connect later than the publisher

---

Current Test Parameters (Modified):

┌───────────────────────┬─────────┬─────────────────────────────────────────────────┐
│ Parameter │ Value │ Notes │
├───────────────────────┼─────────┼─────────────────────────────────────────────────┤
│ iterations │ 1 │ Reduced from original 1,000 for faster feedback │
├───────────────────────┼─────────┼─────────────────────────────────────────────────┤
│ minDelayMs │ 0 │ Startup delay can begin at 0 ms │
├───────────────────────┼─────────┼─────────────────────────────────────────────────┤
│ maxDelayMs │ 1_000 │ Max 1 second random delay (original: 10s) │
├───────────────────────┼─────────┼─────────────────────────────────────────────────┤
│ publishIntervalMs │ 100 │ Publishes every 100 ms (original: 1,000 ms) │
├───────────────────────┼─────────┼─────────────────────────────────────────────────┤
│ iterationTimeout │ 1.1s │ Maximum wait for messages to arrive │
└───────────────────────┴─────────┴─────────────────────────────────────────────────┘

---

Test Execution Flow (Per Iteration):

1. Initialization:

• Find an available TCP port
• Generate 3 independent random delays (0-1,000 ms):
  - Publisher startup delay
  - ALPHA client startup delay
  - BETA client startup delay
• Create TaskCompletionSource instances (alphaReceived, betaReceived)

2. Parallel Startup (3 Async Tasks):

Publisher Task:

• Waits for publisherDelayMs
• Creates EventPublisher instance (with Moq-ed Test Logger)
• Binds to ZeroMQ socket
• Publishes "General" topic messages every 100 ms
• Runs until CancellationToken is triggered

ALPHA Client Task:

• Waits for alphaDelayMs
• Creates ClientPublishChannel instance
• Event Handler: when "General" message arrives → alphaReceived.TrySetResult(true)
• Subscribes to "General" topic
• Starts listening loop in background Task

BETA Client Task:

• Same as ALPHA, with separate topic subscription and event

3. Wait & Cleanup:

allGot = Task.WhenAll(alphaReceived.Task, betaReceived.Task)
.Wait(iterationTimeout); // Max 1.1 seconds

• If both clients received at least 1 "General" message → iteration passes
• Finally block: cancel token, wait for task completion, dispose resources