Update InMemoryScheduler and ScheduledSend

[thedonmon]

So I have been working on adding NATS as a transport protocol for Wolverine and while I was running through the compliance tests before opening the PR, discovered a couple of issues. Would love any feedback or thoughts on this fix but the core of the issue is outlined below.

Summary

Scheduled messages fail to reach their destination when using transports that don't support native scheduled send (RabbitMQ, Kafka, Redis, AWS SQS/SNS, GCP Pub/Sub, MQTT, etc.) in inline or buffered mode. Instead, the message gets received back on the sender's response queue.

Discovery

While implementing a new NATS transport, the schedule_send compliance test was failing. The scheduled message was being sent, but it arrived on the sender's own response subject instead of the receiver's subject.

The bug

For transports without native scheduling, Wolverine wraps the original envelope inside a "scheduled-envelope" wrapper and sends it to local://durable/. When the scheduled time arrives, ScheduledSendEnvelopeHandler unwraps the inner envelope and forwards it to the actual transport.

The problem: the inner envelope is deserialized with its original Status = Scheduled and ScheduledTime still set. Later, when FlushOutgoingMessagesAsync() processes outgoing messages, it calls IsScheduledForLater():

public bool IsScheduledForLater(DateTimeOffset utcNow)
{
    if (Status == EnvelopeStatus.Scheduled) return true;  // <-- 
    return ScheduledTime.HasValue && ScheduledTime.Value > utcNow;
}

Because Status == Scheduled, this returns true even though we're past the scheduled time and should be sending now. This causes the message to skip the actual send path and instead get re-scheduled locally, where it eventually ends up on the wrong queue.

The fix

ScheduledSendEnvelopeHandler.cs

When the scheduled envelope wrapper is processed and we extract the inner envelope, we need to clear the scheduling properties since we're now executing the send:

var scheduled = (Envelope)context.Envelope!.Message!;
scheduled.Source = context.Runtime.Options.ServiceName;
scheduled.ScheduledTime = null;              // We're executing now
scheduled.Status = EnvelopeStatus.Outgoing;  // Ready to send

Without this, IsScheduledForLater() returns true because it checks if (Status == EnvelopeStatus.Scheduled) return true;, causing the message to take the wrong code path in FlushOutgoingMessagesAsync().

InMemoryScheduledJobProcessor.cs

While investigating, we also found two issues in the in-memory scheduler:

1. Wrong time zone: Was using DateTimeOffset.Now instead of DateTimeOffset.UtcNow when calculating delay, but ExecutionTime is stored in UTC
2. Negative delay handling: If execution time had already passed, Task.Delay() would receive a negative TimeSpan

var delayTime = ExecutionTime.Subtract(DateTimeOffset.UtcNow);

if (delayTime <= TimeSpan.Zero)
{
    _task = Task.Run(() => publish());
}
else
{
    _task = Task.Delay(delayTime, _cancellation.Token)
        .ContinueWith(_ => publish(), TaskScheduler.Default);
}

Affected transports

All transports where SupportsNativeScheduledSend returns false:

Testing

Verified with NATS transport compliance tests - schedule_send now passes consistently.

[jeremydmiller]

@thedonmon Thank you for tracking that down! I'm still slow for a couple days, but I'll get this in shortly.

And yell if you need anything for the NATS.io transport. I'd hope there's enough samples by now to make that feasible.

[thedonmon]

@thedonmon Thank you for tracking that down! I'm still slow for a couple days, but I'll get this in shortly.

And yell if you need anything for the NATS.io transport. I'd hope there's enough samples by now to make that feasible.

No problem! And almost complete with the implementation, but I just wanted to verify with all the tests that its behaving correctly! I know its the holidays but I had a little free time to dig into a couple of things 😄